import os import json from pathlib import Path from typing import Annotated from autogen import AssistantAgent, UserProxyAgent from autogen.coding import LocalCommandLineCodeExecutor import gradio as gr from autogen import ConversableAgent from autogen import register_function import mysql.connector import random import requests from groq import Groq from dotenv import load_dotenv tool_resp = "" js = """ function createGradioAnimation() { var container = document.createElement('div'); container.id = 'gradio-animation'; container.style.fontSize = '2em'; container.style.fontWeight = 'bold'; container.style.textAlign = 'center'; container.style.marginBottom = '20px'; var text = '部門収益分析'; for (var i = 0; i < text.length; i++) { (function(i){ setTimeout(function(){ var letter = document.createElement('span'); var randomColor = "#" + Math.floor(Math.random() * 16777215).toString(16); letter.style.color = randomColor; letter.style.opacity = '0'; letter.style.transition = 'opacity 0.5s'; letter.innerText = text[i]; container.appendChild(letter); setTimeout(function() { letter.style.opacity = '1'; }, 50); // Blink the text 3 times for (var j = 0; j < 3; j++) { setTimeout(function() { letter.style.opacity = '0'; }, 500 + j * 1000); setTimeout(function() { letter.style.opacity = '1'; }, 1000 + j * 1000); } }, i * 250); })(i); } var gradioContainer = document.querySelector('.gradio-container'); gradioContainer.insertBefore(container, gradioContainer.firstChild); return 'Animation created'; } """ load_dotenv(verbose=True) conn = mysql.connector.connect( host=os.environ.get("HOST"), user=os.environ.get("USER_NAME"), password=os.environ.get("PASSWORD"), port=os.environ.get("PORT"), database=os.environ.get("DB"), ssl_disabled=True, connection_timeout=60, use_pure=True ) cursor = conn.cursor(dictionary=True) def get_rounrobin(): select_one_data_query = "select api from agentic_apis_count order by counts ASC" cursor.execute(select_one_data_query) result = cursor.fetchall() first_api = result[0]['api'] return first_api # MySQLに接続 def get_api_keys(): token = get_rounrobin() os.environ["GROQ_API_KEY"] = token return token get_api_keys() # Configure Groq config_list = [{ "model": "llama-3.3-70b-versatile", "api_key": os.environ["GROQ_API_KEY"], "api_type": "groq" }] # Create a directory to store code files from code executor work_dir = Path("coding") work_dir.mkdir(exist_ok=True) code_executor = LocalCommandLineCodeExecutor(work_dir=work_dir) # Define revenue tool #def get_current_revenue(location, unit="yen"): def get_current_revenue(location): """Get the revenue for some location""" data = requests.get('https://www.ryhintl.com/dbjson/getjson?sqlcmd=select `title` as country,`snippet` as revenue from cohere_documents_auto') # 元のデータ data = json.loads(data.content) # 指定された形式に変換 revenue_data = {item["country"]: {"revenue": item["revenue"]} for item in data} #print("revenue data:",revenue_data) tmp = json.dumps({ "location": location.title(), "revenue": revenue_data[location]["revenue"], "unit": "" #"unit": unit }) #print("tmp:",tmp) return json.dumps({ "location": location.title(), "revenue": revenue_data[location]["revenue"], "unit": "" }) #return json.dumps({"location": location, "revenue": "unknown"}) # Create an AI assistant that uses the kpi tool assistant = AssistantAgent( #assistant = ConversableAgent( name="groq_assistant", system_message="""あなたは、次のことができる役に立つAIアシスタントです。 - 情報検索ツールを使用する - 結果を分析して自然言語のみで説明する""", llm_config={"config_list": config_list} ) # Create a user proxy agent that only handles code execution user_proxy = UserProxyAgent( #user_proxy = ConversableAgent( name="user_proxy", human_input_mode="NEVER", code_execution_config={"work_dir":"coding", "use_docker":False}, max_consecutive_auto_reply=2, #llm_config={"config_list": config_list} ) '''user_proxy.register_function( function_map={ "get_current_revenue": get_current_revenue } )''' # Register weather tool with the assistant @user_proxy.register_for_execution() @assistant.register_for_llm(description="snippetの内容") #@user_proxy.register_for_llm(description="Weather forecast for cities.") def revenue_analysis( location: Annotated[str, "title"] #unit: Annotated[str, "Revenue unit (dollar/yen)"] = "yen" ) -> str: #revenue_details = get_current_revenue(location=location, unit=unit) revenue_details = get_current_revenue(location=location) revenues = json.loads(revenue_details) #print("resp:",f"{revenues['location']}の内容は{revenues['revenue']}") global tool_resp tool_resp = tool_resp + f"\n\n{location}\n{revenues['location']}の内容は{revenues['revenue']}" return f"{revenues['location']}の内容は{revenues['revenue']}" def get_revenue_and_plot(div1, div2, div3): get_api_keys() # Start the conversation resp = user_proxy.initiate_chat( assistant, message=f"""3つのことをやってみましょう: 1. {div1}、{div2}、{div3}の内容をtoolを利用して抽出します。 2. toolを利用して抽出された内容を詳しく分析します。 3. 日本語で説明してください。 """ ) total_tokens = resp.cost['usage_including_cached_inference']['llama-3.3-70b-versatile']['total_tokens'] #update counts select_one_data_query = "SELECT counts FROM agentic_apis_count where api = '"+os.environ["GROQ_API_KEY"]+"'" cursor.execute(select_one_data_query) ext_key = cursor.fetchall() key = [item['counts'] for item in ext_key] calculated = key[0]+total_tokens/10000 update_counts_query = "UPDATE agentic_apis_count SET counts = "+str(calculated)+" WHERE api = '"+os.environ["GROQ_API_KEY"]+"'" cursor.execute(update_counts_query) conn.commit() groq_assistant_contents = [entry['content'] for entry in resp.chat_history if entry['role'] == 'user' and entry['name'] == 'groq_assistant'] global tool_resp client = Groq(api_key=os.environ["GROQ_API_KEY"]) system_prompt = { "role": "system", "content": "You are a helpful assistant, answer questions concisely." } # Set the user prompt user_input = tool_resp+"を要約してください。" user_prompt = { "role": "user", "content": user_input } # Initialize the chat history chat_history = [system_prompt, user_prompt] response = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=chat_history, max_tokens=1024, temperature=0) kekka = response.choices[0].message.content usages = "使用トークン数: "+str(total_tokens)+ " \n"+kekka return groq_assistant_contents,usages # Create Gradio interface iface = gr.Interface( js=js, fn=get_revenue_and_plot, inputs=[gr.Dropdown(choices=["上期経営会議議事録", "セキュリティー会議資料", "コーポレートガバナンス会議資料"], label="上期経営会議議事録を要約する",value="上期経営会議議事録"), gr.Dropdown(choices=["上期経営会議議事録", "セキュリティー会議資料", "コーポレートガバナンス会議資料"], label="セキュリティー会議資料を要約する",value="セキュリティー会議資料"), gr.Dropdown(choices=["上期経営会議議事録", "セキュリティー会議資料", "コーポレートガバナンス会議資料"], label="コーポレートガバナンス会議資料を要約する",value="コーポレートガバナンス会議資料")], outputs=[gr.Textbox(label="結果"),gr.Textbox(label="Usageデータとツール結果")], title="資料の分析（AUTOGEN）", description="プロンプトを入力してデータを取得し、内容を分析します。", submit_btn="実行", clear_btn="クリア", flagging_mode="never" ) iface.launch()