څنګه AI د Confluence & Streamlit سره د چټک پروګرام مدیریت بدلون کوي

`Problem Statement:`

د ستونزو اعلان:

د Copilot سره په سازمانونو غوښتنلیکونو کې انډول شوی، د فایلونو، SharePoint، او نورو لاس رڼاونو څخه نږدې کارول شوي ډاټا په لټه کې کولی شي. زه ډیر د دې Gen AI وړتیا پرته له لاسه ورکړم. يو ورځ، زه د ټولو ځانګړتیاوو (د ټیم په Agile Framework کې د کټګورۍ لپاره د کټګورۍ وړتیاوې) او د دوی د حالتونو په اړه یو خلاص نظر ته اړتيا لري چې د ټیم کار کوي. ناڅاپي، Copilot د Confluence پاڼه څخه د معلوماتو په لټه کې راځي، کوم چې مناسب او انتظار دی. ډیری سازمانونه، پروژې او پروګرامونه د Confluence پاڼه کې ذخیره شوي دي. د ټیم هدفونه، deliverables، پروژې خطرونه

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

د حل: د AI Agentic Assistant له خوا Streamlit + Semantic Kernel Powered

د ایجنټیک AI نندارتون زما لپاره یو خوندي دی، او زه د دې فریم ورک په توګه د حل په توګه انتخاب وکړم. په هرصورت، ستونزه شتون نلري: کوم فریم ورک باید کارول شي، او آیا یو خوندي سرچينې شتون لري؟ د مدیریت شوي پلیټ فارم چقدر ارزانه وي؟ په پایله کې، د ټولو څیړنو سره، زه د خوندي سرچينې سره ونیسئ او د لاندې تکنالوژۍ کڅوړه کارولو لپاره د یو رڼا AI مسلک جوړ کړي چې د کارولو سره کاروي:

Streamlit د مخ په پای کې،
Semantic Kernel لپاره د چټک مدیریت او چټکولو،
Azure OpenAI د طبیعي زبان پروسس لپاره
د Confluence سایټونو د خوندي او ډيزاينیک ویب scraping لپاره.

هغه څه چې

دا وسیله به د پروګرام مدیرانو او رهبريانو ته اجازه ورکوي:

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`

څنګه کار کوي - Pseudo Code

Step 1. Confluence Page LookUp.

د URLs په کارولو په ځای کې، هر ټیم نوم د یو لغتار په کارولو سره په خپل Confluence URL کې کارول کیږي. کله چې یو کاروونکي د "Team A" څخه د راستې پینل څخه انتخاب کوي، د بکسین په اتوماتيک ډول د تړاو URL راټول کوي او د scraping فعالوي.

team_to_url_map = { "د ټیم A": "https://confluence.company.com/display/TEAM_A"،

"د ټیم بی": "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

په پای کې، زه د پلورونکي پر بنسټ د سر پر بنسټ د scraping لپاره Playwright کارول، کوم چې موږ سره مرسته کوي چې د دومايښت موادو لوستل او د لاس ریکارډونو په کارولو کې مرسته وکړي:

ناڅاپي لارښوونې:

[ ]Python غوښتنلیک کتابتون په کارولو سره، د API په کارولو سره د Confluence ډاټا ترلاسه کړئ. د تصدیق ميخانيکي بریالیتوب نلري. په بل ډول، دا به د Confluence پاڼه ډاټا ترلاسه کولو یو ښه لاره وي.
[ ]Python BeautifySoup کتابتون کاروي. دا د ډینیمي موادو له امله غیر فعال شوی.
[ ] زه د Python Playwright سره پای ته ورسیږي. د SSO کچه ستونزه لري، مګر په پای کې، دا د HTML حالت JSON ډاونلوډ کولو او بیا کارولو وروسته کار کړ.

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

@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.

دا وسیله د پروګرام مدیریت فعالولو لپاره دی. د ایجنټ لارښوونې د سپریټ موادو کارولو لپاره جوړ شوي دي او د PM پوښتنو لپاره مناسب پايلې تولیدوي. لارښوونې به د ټیکټ خلاص یا د چارټ فارم په توګه د پایلو ترلاسه کولو کې مرسته وکړي. دا هم د ټیټ کوډ مثال دی.

برسېره پر دې، زه د کلینټ د AI Agent په توګه د پلگ ان په توګه تعریف کړ.

AGENT_INSTRUCTIONS = “”

د

کلینټر = OpenAI(<د محلي Open Source LLM>)

chat_completion_service = OpenAIChatCompletion(ai_model_id="<>"،

async_client = مشتری )

agent = ChatCompletionAgent( service=chat_completion_service، پلگ انز=[ ConfluencePlugin() ]، نام="ConfluenceAgent", لارښوونې=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.

زه د اضافي LLM کلینر اضافه کولو لپاره ټاکل شوي ترڅو چمتو کړي که آیا د ګرځنده وارداتو د پروګرام مدیریت لپاره مهم دی یا نه.

completion = await client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": " تاسو د کاروونکي د وارداتو د موادو په اړه داست. د کاروونکي د وارداتو په اړه داست. که دا د يو ټیم لپاره د داخلي COnfluence پاڼه راټولولو ته اړتیا لري نو دا د پروګرام مدیریت سره اړتیا لري. که دا د پروګرام مدیریت سره اړتیا نلري، د ځواب ته ځواب اضافه کړئ مګر د ځواب ته 'Falseḳ' اضافه کړئ. که دا د پروګرام مدیریت سره اړتیا لري، د ځواب ته 'Trueḳ' اضافه کړئ."}، {"role": "user", "content": user_input} ]، د حرارت درجه=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.

دلته ټول کوډ دی. زه زما د پروژې ځانګړي تفصيلات حذف کړ. موږ باید لومړی د 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 matplotlibot.pyplot as plt from semantic_kernel.agents import ChatCompletionAgent from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion from semantic_kernel.contents import FunCallContent, FunctionResultContent, StreamingTextContent from semantic_kern

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" }

# ---- د پلگ ان تعریف ----

#د بڼه د فریکونسۍ ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډ

def extract_json_from_response(text): # د regex کارولو سره په متن کې د لومړي JSON نندارتون ته وټاکئ = re.search(r"(\[\s*{.*\s*\])"، text، re.DOTALL) که سره وټاکئ: د match.group(1) ته ورسیږي نه

ټولګي ConfluencePlugin: def init(self): self.default_confluence_url = "<>" load_dotenv()
@kernel_function(description="Scrape and return text content from a Confluence page.")
async def get_confluence_page_content(
            self, team_name: Annotated\[str, "Name of the Agile team"\]
            ) -> Annotated\[str, "Returns extracted text content from the page"\]:
            print(f"Attempting to scrape Confluence page for team: '{team_name}'") # Added for debugging
            target_url = TEAM_URL_MAPPING.get(team_name)
            if not target_url:
                print(f"Failed to find URL for team: '{team_name}' in TEAM_URL_MAPPING.") # Added for debugging
                return f"❌ No Confluence URL mapped for team '{team_name}'"

            async with async_playwright() as p:
                browser = await p.chromium.launch()
                context = await browser.new_context(storage_state="state.json")
                page = await context.new_page()                    
                pages_to_scrape = \[target_url\]

                # Loop through each page URL and scrape the content
                for page_url in pages_to_scrape:
                    await page.goto(page_url)
                    await asyncio.sleep(30)  # Wait for the page to load
                    await page.wait_for_selector('div.refresh-module-id, table.some-jira-table')
                    html = await page.content()
                    soup = BeautifulSoup(html, "html.parser")
                    body_div = soup.find("div", class_="wiki-content") or soup.body
                    if not body_div:
                        return "❌ Could not find content on the Confluence page."
                    # Process the scraped content (example: extract headings)
                    headings = soup.find_all('h2')
                    text = body_div.get_text(separator="\\n", strip=True)
                    return text\[:4000\]  # Truncate if needed to stay within token limits
                await browser.close()               
@kernel_function(description="Summarize and structure scraped Confluence content into JSON.")
async def summarize_confluence_data(
    self, raw_text: Annotated\[str, "Raw text scraped from the Confluence page"\],
    output_style: Annotated\[str, "Output style, either 'bullet' or 'json'"\] = "json"  # Default to 'json'
) -> Annotated\[str, "Returns structured summary in JSON format"\]:

    prompt = f"""
    You are a Program Management Data Extractor.
    Your job is to analyze the following Confluence content and produce structured machine-readable output.

    Confluence Content:
    {raw_text}

    Instructions:
    - If output_style is 'bullet', return bullet points summary.
    - If output_style is 'json', return only valid JSON array by removing un printable characters and spaces from beginning and end.
    - DO NOT write explanations.
    - DO NOT suggest code snippets.
    - DO NOT wrap JSON inside triple backticks \`\`\`json
    - Output ONLY the pure JSON array or bullet points list.

    Output_style: {output_style}
        """

    # Call OpenAI again
    completion = await client.chat.completions.create(
        model="gpt-4o",
        messages=\[
            {"role": "system", "content": "You are a helpful Program Management Data Extractor."},
            {"role": "user", "content": prompt}
        \],
        temperature=0.1
        )

    structured_json = completion.choices\[0\].message.content.strip()
    return structured_json
# ---- Load API اعتبارات ---- load_dotenv() کلینټ = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client )

AGENT_INSTRUCTIONS = """ تاسو یو ګټور پروګرام مدیریت AI ایجنټ دی چې کولی شئ د یو ګډې پاڼه څخه د مهمو معلوماتو لکه ټیم ارقام، ځانګړتیاوې، Epics اخلي.

مهم: کله چې کاروونکي د ټیم پاڼه مشخصوي، یوازې د دې ټیم د ځانګړتیاوو او Epics اخلي.

کله چې د بحث پیل کیږي، خپل ځان د دې پیغام سره وړاندې کړئ: "د سلام! زه ستاسو د PM مسلکي. زه کولی شئ تاسو ته د ځانګړتیاوو او Epics حالت ترلاسه کړي.

لارښوونه چې تاسو باید پیژندل کړئ: 1. تل لومړی د 'get_confluence_page_content' ته وښيي ترڅو د Confluence پاڼه راټول کړئ.

که د کاروونکي پیغام د "Team: {team_name}." سره پیل کیږي، دا {team_name} د "team_name" انډول لپاره کاروي. د مثال په توګه، که د انډول "Team: Raptor. د وروستیو ځانګړتیاوو په توګه څه دي؟"، د "team_name" د "Raptor" دی. 2. که کاروونکي د خلاصې لپاره پوښتنه کوي، د توپ ټکي لیست وړاندې کړئ. 3. که کاروونکي د JSON سیسټم یا ټکي ټکي یا ټکي ته پوښتنه کوي. نو فورا د 'summarize_confluence_data' له لارې د سپیک شوي محتویاتو په کارولو سره تماسئ. 4. د کاروونکي لخوا غوښتل شوي پایلې سبک په اساس، یو JSON سیسټم یا توپ ټکي ته ورسئ. 5. که کاروونکي

لارښوونه: - که output_style د 'Bullet' دی، د Bullet ټکي خلاص ته راځي. - که output_style د 'JSON' دی، یوازې د بریښنا وړ JSON نښانې ته راځي د پیل او پای څخه د چاپ وړ نښانونه او بڼهونو حذف کولو له لارې. - د تفسیرونو لیکنه نه. - د کوډ نندارتونونه نندارې. - د JSON په ټریپل backticks '`JSON کې نندارې. - یوازې د خالص JSON نښانې یا Bullet ټکي لیست صادر کړئ.

تاسو په کوم ټیم کې غواړئ چې تاسو ته د نن ورځ په پلان کې مرسته وکړي؟"

په هر وخت کې د کاروونکي ترټولو ترټولو ترټولو غوره کړئ. که دوی د یو ځانګړي ټیم په اړه اشاره کوي، نو ستاسو ډاټا په دې ټیم کې تمرکز وکړئ او نه د بدیلونو وړاندیز وکړئ. """ agent = ChatCompletionAgent( service=chat_completion_service، plugins=[ ConfluencePlugin() ]، name="ConfluenceAgent", instructions=AGENT_INSTRUCTIONS )

# ---- Main async logic ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = نه بشپړولو لپاره د يو ټیم لپاره د کورني COnfluence پاڼه پوښتنه = انتظار client.chat.completions.create( نمونې="gpt-4o"، پیغامونه=["د سیسټم"، "تقسيم" ته ځواب اضافه کړئ. که دا د پروګرام مدیریت سره تړاو لري، په ځواب کې د 'True Fitt' اضافه کړئ.}،" {role": "تقسيم"، "تقسيم"، "تقسيم": د پروژې د پروژې د حرارت درجه، text=0.5 ) ځواب =_text_cho
    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, بدن, [class*="css"] { font-size: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- سره left_col: st.title(" پروګرام مدیریت Enabler AI") st.write("د مختلفو ویلی پروګرامونه په اړه پوښتنه!") 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"\])
with right_col:st.title("د ویلی پروګرام انتخاب کړئ")
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 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 matplotlibot.pyplot as plt from semantic_kernel.agents import ChatCompletionAgent from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion from semantic_kernel.contents import FunCallContent, FunctionResultContent, StreamingTextContent from semantic_kern

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" }

# ---- د پلگ ان تعریف ----

#د بڼه د فریکونسۍ ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډیزاین ډ

def extract_json_from_response(text): # د regex کارولو سره په متن کې د لومړي JSON نندارتون ته وټاکئ = re.search(r"(\[\s*{.*\s*\])"، text، re.DOTALL) که سره وټاکئ: د match.group(1) ته ورسیږي نه

ټولګي ConfluencePlugin: def init(self): self.default_confluence_url = "<>" load_dotenv()

@kernel_function(description="Scrape and return text content from a Confluence page.")
async def get_confluence_page_content(
            self, team_name: Annotated\[str, "Name of the Agile team"\]
            ) -> Annotated\[str, "Returns extracted text content from the page"\]:
            print(f"Attempting to scrape Confluence page for team: '{team_name}'") # Added for debugging
            target_url = TEAM_URL_MAPPING.get(team_name)
            if not target_url:
                print(f"Failed to find URL for team: '{team_name}' in TEAM_URL_MAPPING.") # Added for debugging
                return f"❌ No Confluence URL mapped for team '{team_name}'"

            async with async_playwright() as p:
                browser = await p.chromium.launch()
                context = await browser.new_context(storage_state="state.json")
                page = await context.new_page()                    
                pages_to_scrape = \[target_url\]

                # Loop through each page URL and scrape the content
                for page_url in pages_to_scrape:
                    await page.goto(page_url)
                    await asyncio.sleep(30)  # Wait for the page to load
                    await page.wait_for_selector('div.refresh-module-id, table.some-jira-table')
                    html = await page.content()
                    soup = BeautifulSoup(html, "html.parser")
                    body_div = soup.find("div", class_="wiki-content") or soup.body
                    if not body_div:
                        return "❌ Could not find content on the Confluence page."
                    # Process the scraped content (example: extract headings)
                    headings = soup.find_all('h2')
                    text = body_div.get_text(separator="\\n", strip=True)
                    return text\[:4000\]  # Truncate if needed to stay within token limits
                await browser.close()               
@kernel_function(description="Summarize and structure scraped Confluence content into JSON.")
async def summarize_confluence_data(
    self, raw_text: Annotated\[str, "Raw text scraped from the Confluence page"\],
    output_style: Annotated\[str, "Output style, either 'bullet' or 'json'"\] = "json"  # Default to 'json'
) -> Annotated\[str, "Returns structured summary in JSON format"\]:

    prompt = f"""
    You are a Program Management Data Extractor.
    Your job is to analyze the following Confluence content and produce structured machine-readable output.

    Confluence Content:
    {raw_text}

    Instructions:
    - If output_style is 'bullet', return bullet points summary.
    - If output_style is 'json', return only valid JSON array by removing un printable characters and spaces from beginning and end.
    - DO NOT write explanations.
    - DO NOT suggest code snippets.
    - DO NOT wrap JSON inside triple backticks \`\`\`json
    - Output ONLY the pure JSON array or bullet points list.

    Output_style: {output_style}
        """

    # Call OpenAI again
    completion = await client.chat.completions.create(
        model="gpt-4o",
        messages=\[
            {"role": "system", "content": "You are a helpful Program Management Data Extractor."},
            {"role": "user", "content": prompt}
        \],
        temperature=0.1
        )

    structured_json = completion.choices\[0\].message.content.strip()
    return structured_json

# ---- Load API اعتبارات ---- load_dotenv() کلینټ = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client )

AGENT_INSTRUCTIONS = """ تاسو یو ګټور پروګرام مدیریت AI ایجنټ دی چې کولی شئ د یو ګډې پاڼه څخه د مهمو معلوماتو لکه ټیم ارقام، ځانګړتیاوې، Epics اخلي.

مهم: کله چې کاروونکي د ټیم پاڼه مشخصوي، یوازې د دې ټیم د ځانګړتیاوو او Epics اخلي.

کله چې د بحث پیل کیږي، خپل ځان د دې پیغام سره وړاندې کړئ: "د سلام! زه ستاسو د PM مسلکي. زه کولی شئ تاسو ته د ځانګړتیاوو او Epics حالت ترلاسه کړي.

لارښوونه چې تاسو باید پیژندل کړئ: 1. تل لومړی د 'get_confluence_page_content' ته وښيي ترڅو د Confluence پاڼه راټول کړئ.

که د کاروونکي پیغام د "Team: {team_name}." سره پیل کیږي، دا {team_name} د "team_name" انډول لپاره کاروي. د مثال په توګه، که د انډول "Team: Raptor. د وروستیو ځانګړتیاوو په توګه څه دي؟"، د "team_name" د "Raptor" دی. 2. که کاروونکي د خلاصې لپاره پوښتنه کوي، د توپ ټکي لیست وړاندې کړئ. 3. که کاروونکي د JSON سیسټم یا ټکي ټکي یا ټکي ته پوښتنه کوي. نو فورا د 'summarize_confluence_data' له لارې د سپیک شوي محتویاتو په کارولو سره تماسئ. 4. د کاروونکي لخوا غوښتل شوي پایلې سبک په اساس، یو JSON سیسټم یا توپ ټکي ته ورسئ. 5. که کاروونکي

لارښوونه: - که output_style د 'Bullet' دی، د Bullet ټکي خلاص ته راځي. - که output_style د 'JSON' دی، یوازې د بریښنا وړ JSON نښانې ته راځي د پیل او پای څخه د چاپ وړ نښانونه او بڼهونو حذف کولو له لارې. - د تفسیرونو لیکنه نه. - د کوډ نندارتونونه نندارې. - د JSON په ټریپل backticks '`JSON کې نندارې. - یوازې د خالص JSON نښانې یا Bullet ټکي لیست صادر کړئ.

تاسو په کوم ټیم کې غواړئ چې تاسو ته د نن ورځ په پلان کې مرسته وکړي؟"

په هر وخت کې د کاروونکي ترټولو ترټولو ترټولو غوره کړئ. که دوی د یو ځانګړي ټیم په اړه اشاره کوي، نو ستاسو ډاټا په دې ټیم کې تمرکز وکړئ او نه د بدیلونو وړاندیز وکړئ. """ agent = ChatCompletionAgent( service=chat_completion_service، plugins=[ ConfluencePlugin() ]، name="ConfluenceAgent", instructions=AGENT_INSTRUCTIONS )

# ---- Main async logic ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = None 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 ) response_text = completion.choices[0].message.content.strip() print("Response Text:", response_text) if response_text.startswith("False|"): response_text = response_text[6:] yield response_text, thread, [] # Return the response without any further processingelif response_text.startswith("True|"): response_text = response_text[5:]function_calls = re.findall(r"(\w+\(.*?\))", response_text) # Remove the function calls from the response text for call in function_calls: response_text = response_text.replace(call, "") # If the response is related to Program Management, continue processing

    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, بدن, [class*="css"] { font-size: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- سره left_col: st.title(" پروګرام مدیریت Enabler AI") st.write("د مختلفو ویلی پروګرامونه په اړه پوښتنه!") 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"\])

with right_col:st.title("د ویلی پروګرام انتخاب کړئ")

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

د پایلو

دStreamlit-based Program Management AI chatbot helps teams track project features and epics from Confluence pages. The application uses Semantic Kernel agentsد OpenAI GPT-4o سره د ټیم ځانګړي Confluence پاڼه محتویاتو کارولو لپارهPlaywright. د حالت د تصدیق لپاره کارول کیږي. د وسیله اجازه ورکوي چې د پروګرام او د اړونده ټیم غوره کړي، او د انتخاب په اساس، د کاروونکي د وارداتو ته ځواب ورکړي. د Agentic AI ځانګړتیاوو سره، موږ کولی شو د LLM لپاره یو حقیقي شخصي مسلکي وي. دا کولی شي د LLM د معلوماتو ته لاس رسی لپاره قوي وي، مګر په هرصورت د محدود معلوماتو د LLM ځانګړتیا څخه ګټه واخلئ. دا د Agentic AI ځانګړتیاوو د درک لپاره یو مثال دی او دا څومره قوي وي.

د نندارتون:https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file

              Playwright documentation.

څنګه AI د Confluence & Streamlit سره د چټک پروګرام مدیریت بدلون کوي

ډېر اوږد؛ لوستل

`Problem Statement:`

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

هغه څه چې

`How it works - Pseudo Code`

د پایلو

About Author

hang tags

دا مقاله په کې وړاندې شوې وه...

Trending Topics

Classic

Neon Noir

Minty

Newspaper

HN StartUps

څنګه AI د Confluence & Streamlit سره د چټک پروګرام مدیریت بدلون کوي

ډېر اوږد؛ لوستل

Problem Statement:

Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel

هغه څه چې

How it works - Pseudo Code

د پایلو

About Author

hang tags

دا مقاله په کې وړاندې شوې وه...

اړوندې کیسې

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`