app.py

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()