Problem Statement: Problemer med udtalelse: Med Copilot integreret i organisationens applikationer er det blevet utroligt nemt at finde sjældent brugte data fra filer, SharePoint og andre tilgængelige kilder. Jeg har været stærkt afhængig af denne Gen AI-funktion. En dag havde jeg brug for en oversigt over alle funktioner (teamets leverancer til et kvartal i Agile Framework) og deres status, som teamet arbejder på. Desværre nægtede Copilot at læse data fra Confluence-siden, hvilket er ideelt og forventet. Mange organisationer, projekter og programopdateringer er gemt på Confluence-sider. At få et overblik over teammål, leverancer, projektrisici og status kan være tidskrævende for en leder eller en person, der håndterer flere programmer. Jeg tænkte, hvorfor ikke have en intelligent assistent til at hente og opsummere Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel Løsning: AI Agent Assistant drevet af Streamlit + Semantic Kernel Introduktionen af Agentic AI var en frelser for mig, og jeg besluttede at vælge dette framework som en løsning. Men der var udfordringer: Hvilket framework skal bruges, og er der en tilgængelig open source? Hvor dyrt ville den administrerede platform være? Endelig besluttede jeg med al forskning at gå med open-source og bruge nedenstående tech stack til at opbygge en let AI assistent ved hjælp af: Strømslys til frontenden, Semantisk kernel til hurtig ledelse og kæde, Azure OpenAI til naturlig sprogbehandling Playwright til sikker og dynamisk webskrabning af Confluence-sider. Hvad det gør Dette værktøj vil gøre det muligt for programledere og ledere at: Select a from a dropdown, team or program name Automatically fetch the associated Confluence page URL, Scrape key content sections from that page (like Features, Epics, Dependencies, Risks, Team Members), Ask questions like “What are the team Q4 deliverables?” or “Summarize the features based on status,” etc., Display answers as summarized text. How it works - Pseudo Code Hvordan det virker - Pseudo Code Step 1. Confluence Page LookUp. I stedet for manuelt at indsætte URL'er, mapperes hvert teamnavn til deres Confluence URL ved hjælp af et ordbog. Når en bruger vælger "Team A" fra højre vindue, henter backend automatisk den tilknyttede URL og udløser skrabning. 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 Endelig sluttede jeg med at bruge Playwright til hovedløs browserbaseret scraping, som hjælper os med at indlæse dynamisk indhold og håndtere login: Den mislykkede tilgang: [ ]Ved hjælp af Python-forespørgselsbiblioteket, få Confluence Data ved hjælp af API. Autentificeringsmekanismen var ikke vellykket. Ellers ville det være en fremragende måde at få Confluence side data. [ ] Brug af Python BeautifySoup-biblioteket. Det blev udelukket på grund af dynamisk indhold. [ ]Jeg endte med Python Playwright. SSO-laget havde udfordringer, men til sidst fungerede det efter at have downloadet HTML-staten JSON og genbrugt det. @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. Værktøjet er beregnet til at være en Program Management Enabler. Agent Instruktionerne er udarbejdet til at bruge det skrabede indhold og producere et resultat egnet til PM spørgsmål. Instruktionerne vil hjælpe med at få output som en tekstoversigt eller i et diagramformat. Dette er også et eksempel på lav kode. Desuden har jeg defineret klienten som en AI Agent som et plugin. Agent_Instruktioner = ”” »« klient = OpenAI (<Local open source LLM>) chat_completion_service = OpenAIChatCompletion(ai_model_id="<>", async_client = bruger agent = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], navn="ConfluenceAgent", instruktioner=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. Jeg besluttede at tilføje en ekstra LLM-klient for at kontrollere, om brugerindtastningen er relevant for Program Management eller ej. completion = await client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "Du er en Dommer over indholdet af brugerindtastning. Anlyse brugerens input. Hvis den beder om at skrabe intern COnfluence-side for et team, så er det relateret til Program Management. Hvis det ikke er relateret til Program Management, giv svaret, men tilføj 'Falseḳ' til svaret. Hvis det er relateret til Program Management, tilføj 'Trueḳ' til svaret."}, {"role": "user", "content": user_input} ], temperatur=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. Her er hele koden. Jeg har slettet mine projektspecifikke detaljer. Vi skal først gemme state.json for at bruge den i koden import json import os import asyncio import pandas som pd import streamlit som st fra typing import Annoteret fra dotenv import load_dotenv fra openai import AsyncAzureOpenAI fra playwright.async_api import async_playwright fra bs4 import BeautifulSoup fra semantic_kernel.functions import kernel_function fra typing import Annoteret import re import matplotlibot.pyplot som plt fra semantic_kernel.agents import ChatCompletionAgent fra semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion fra semantic_kernel.contents import FunCallContent, FunctionResultContent, StreamingTextContent fra 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" } # ---- Plugin definition ---- #Bardiagram med fast størrelse def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # bredde, højde i inches ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Skift tick farve ax.tick_params(axis='x', colors='blue', labelrotation=90) # x-ticks i blå, roteret axtick_params(axis='y', colors='green') # y-ticks i grøn st.pyplot(fig) def extract_json_from_response(tekst): # Brug regex til at finde det første JSON-array i teksten match = re.search(r"(\[\s*{.*}\s*\)", tekst, re.DOTALL) hvis match: return match.group(1) return Ingen klasse ConfluencePlugin: def init(selv): 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 credentials ---- 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 = """Du er en nyttig Program Management AI Agent, der kan hjælpe med at udtrække vigtige oplysninger som Team Medlem, Funktioner, Epics fra en sammenflydelsesside. Vigtigt: Når brugerne angiver en teamside, skal de kun udtrække funktionerne og epikerne for det pågældende team. Når samtalen begynder, skal du præsentere dig selv med denne besked: "Hej! jeg er din PM assistent. Først skal du altid ringe til 'get_confluence_page_content' for at skrabe Confluence-siden. Hvis brugerens meddelelse starter med "Team: {team_name}.", skal du bruge det {team_name} til argumentet 'team_name'. For eksempel, hvis indtastningen er "Team: Raptor. Hvad er de nyeste funktioner?", er 'team_name' "Raptor". 2. Hvis brugeren anmoder om en opsummering, skal du angive en liste over kuglepunkter. 3. Hvis brugeren anmoder om et JSON-array eller -diagram eller -diagram. Så skal du straks ringe til 'summarize_confluence_data' ved hjælp af det skrapede indhold. 4. Baseret på den udgangsstil, som brugeren anmoder om, skal du returnere enten et JSON-array eller et kuglepunkt. 5. Hvis brugeren ikke angiver en udgangsstil, skal du Instruktioner: - Hvis output_style er 'bullet', return bullet points summary. - Hvis output_style er 'json', return kun gyldig JSON array ved at fjerne un printbare tegn og mellemrum fra begyndelsen og slutningen. - SKRIV IKKE forklaringer. - SKRIV IKKE foreslå kode snippets. - SKRIV IKKE JSON inde i triple backticks ```json - Output KUN den rene JSON array eller bullet points liste. Hvilket team er du interesseret i at hjælpe dig med at planlægge i dag?" Hvis de nævner et bestemt team, fokuser dine data på det team i stedet for at foreslå alternativer. """ agent = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], name="ConfluenceAgent", instruktioner=AGENT_INSTRUCTIONS ) # ---- Main async logic ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = None completion = wait client.chat.completions.create( model="gpt-4o", messages=["role": "system", "content": "Du er en dommer af indholdet af bruger input. Anlyze brugerens funktion. Hvis det beder om at skrabe intern COnfluence Page for et team, så er det relateret til Program Management. Hvis det ikke er relateret til Program Management, angive svaret, men tilføje 'False Tilt' til svaret. Hvis det er relateret til Program Management, tilføje 'True Tilt' til 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, krop, [class*="css"] { font-størrelse: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- med left_col: st.title(" Program Management Enabler AI") st.write("Spørg mig om forskellige Wiley Program forpligtede elementer.!") st.write("Jeg kan hjælpe dig med at få status for Funktioner og 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"\]) med 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 som pd import streamlit som st fra typing import Annoteret fra dotenv import load_dotenv fra openai import AsyncAzureOpenAI fra playwright.async_api import async_playwright fra bs4 import BeautifulSoup fra semantic_kernel.functions import kernel_function fra typing import Annoteret import re import matplotlibot.pyplot som plt fra semantic_kernel.agents import ChatCompletionAgent fra semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion fra semantic_kernel.contents import FunCallContent, FunctionResultContent, StreamingTextContent fra 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" } # ---- Plugin definition ---- #Bardiagram med fast størrelse def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # bredde, højde i inches ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Skift tick farve ax.tick_params(axis='x', colors='blue', labelrotation=90) # x-ticks i blå, roteret axtick_params(axis='y', colors='green') # y-ticks i grøn st.pyplot(fig) def extract_json_from_response(tekst): # Brug regex til at finde det første JSON-array i teksten match = re.search(r"(\[\s*{.*}\s*\)", tekst, re.DOTALL) hvis match: return match.group(1) return Ingen klasse ConfluencePlugin: def init(selv): 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 credentials ---- 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 = """Du er en nyttig Program Management AI Agent, der kan hjælpe med at udtrække vigtige oplysninger som Team Medlem, Funktioner, Epics fra en sammenflydelsesside. Vigtigt: Når brugerne angiver en teamside, skal de kun udtrække funktionerne og epikerne for det pågældende team. Når samtalen begynder, skal du præsentere dig selv med denne besked: "Hej! jeg er din PM assistent. Først skal du altid ringe til 'get_confluence_page_content' for at skrabe Confluence-siden. Hvis brugerens meddelelse starter med "Team: {team_name}.", skal du bruge det {team_name} til argumentet 'team_name'. For eksempel, hvis indtastningen er "Team: Raptor. Hvad er de nyeste funktioner?", er 'team_name' "Raptor". 2. Hvis brugeren anmoder om en opsummering, skal du angive en liste over kuglepunkter. 3. Hvis brugeren anmoder om et JSON-array eller -diagram eller -diagram. Så skal du straks ringe til 'summarize_confluence_data' ved hjælp af det skrapede indhold. 4. Baseret på den udgangsstil, som brugeren anmoder om, skal du returnere enten et JSON-array eller et kuglepunkt. 5. Hvis brugeren ikke angiver en udgangsstil, skal du Instruktioner: - Hvis output_style er 'bullet', return bullet points summary. - Hvis output_style er 'json', return kun gyldig JSON array ved at fjerne un printbare tegn og mellemrum fra begyndelsen og slutningen. - SKRIV IKKE forklaringer. - SKRIV IKKE foreslå kode snippets. - SKRIV IKKE JSON inde i triple backticks ```json - Output KUN den rene JSON array eller bullet points liste. Hvilket team er du interesseret i at hjælpe dig med at planlægge i dag?" Hvis de nævner et bestemt team, fokuser dine data på det team i stedet for at foreslå alternativer. """ agent = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], name="ConfluenceAgent", instruktioner=AGENT_INSTRUCTIONS ) # ---- Main async logic ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = None completion = wait client.chat.completions.create( model="gpt-4o", messages=["role": "system", "content": "Du er en dommer af indholdet af bruger input. Anlyze brugerens funktion. Hvis det beder om at skrabe intern COnfluence Page for et team, så er det relateret til Program Management. Hvis det ikke er relateret til Program Management, angive svaret, men tilføje 'False Tilt' til svaret. Hvis det er relateret til Program Management, tilføje 'True Tilt' til 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, krop, [class*="css"] { font-størrelse: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- med left_col: st.title(" Program Management Enabler AI") st.write("Spørg mig om forskellige Wiley Program forpligtede elementer.!") st.write("Jeg kan hjælpe dig med at få status for Funktioner og 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"\]) med 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 Konklusionen Den hjælper teams med at spore projektfunktioner og epik fra Confluence-sider. med OpenAI GPT-4o til at skrabe team-specifikke Confluence-sideindhold ved hjælp af Staten bruges til godkendelse. Værktøjet giver mulighed for udvælgelse af programmet og det relevante team, og baseret på udvælgelsen vil brugerens input blive besvaret. Med agent AI-funktionen kan vi gøre det muligt for LLM at være en reel personlig assistent. Det kan være kraftfuldt i at begrænse LLM's adgang til dataene, men stadig udnytte LLM-funktionen af begrænsede data. Det er et eksempel på at forstå agent AI-funktionen og hvor kraftfuld det kan være. Streamlit-based Program Management AI chatbot Semantic Kernel agents Playwright af Reference: https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file Playwright documentation.
