import os import gradio as gr from autogen import AssistantAgent, UserProxyAgent, config_list_from_json, tools from autogen.cache import Cache from autogen.coding import DockerCommandLineCodeExecutor, LocalCommandLineCodeExecutor import mysql.connector from typing import Literal from typing_extensions import Annotated import random import groq from dotenv import load_dotenv load_dotenv(verbose=True) conn = mysql.connector.connect( host="www.ryhintl.com", user="smairuser", password="smairuser", port=36000, database="smair" ) cursor = conn.cursor(dictionary=True) # MySQLに接続 '''def get_api_keys(): select_one_data_query = "SELECT * FROM agentic_apis_copy where api = 'GROQ_KEYS'" cursor.execute(select_one_data_query) result = cursor.fetchall() keys = [item['key'] for item in result] rtnkey = ",".join(map(str, keys)) return rtnkey # List of Groq tokens mytokens = get_api_keys() tokens = eval("["+mytokens+"]") client = groq.Client(api_key=tokens[0])''' client = groq.Client(api_key=os.environ["GROQ_API_KEY"]) def compare_result(eprag: str, llm: str): completion = client.chat.completions.create( model="openai/gpt-oss-120b", messages=[ {"role": "system", "content": "貴方は優秀なアシスタントです。必ず、日本語で答えてください。"}, {"role": "user", "content": eprag+"\n"+llm+"\n epragとllmの結果を比較分析してください。必ず、日本語で答えてください。"} ], ) return completion.choices[0].message.content def get_llm(prompt: str): completion = client.chat.completions.create( model="openai/gpt-oss-120b", messages=[ {"role": "system", "content": "貴方は優秀なアシスタントです。必ず、日本語で答えてください。"}, {"role": "user", "content": prompt} ], ) return completion.choices[0].message.content '''def get_next_token(): token = tokens[random.randint(0, len(tokens) - 1)] return token token = get_next_token()''' token = os.environ["GROQ_API_KEY"] config_list = [ {"model": "llama-3.3-70b-versatile", "api_key": token, "api_type": "groq"}, {"model": "llama3-70b-8192", "api_key": token, "api_type": "groq"}, ] os.makedirs("coding", exist_ok=True) code_executor = LocalCommandLineCodeExecutor(work_dir="coding") user_proxy = UserProxyAgent( name="user_proxy", is_termination_msg=lambda x: x.get("content", "") and x.get("content", "").rstrip().endswith("TERMINATE"), human_input_mode="TERMINATE", max_consecutive_auto_reply=10, code_execution_config={"executor": code_executor}, ) writing_assistant = AssistantAgent( name="writing_assistant", #system_message="あなたは、魅力的なブログ記事を書くことを任されたライティング アシスタントです。ユーザーのリクエストにできる限り応えられる最高のブログ記事を具体的な例による深みとリアリティを交えながら作成しようとします。ユーザーから批判があった場合は、以前の試みを修正したバージョンで応答します。", system_message='''あなたは、魅力的なブログ記事を書くことを任されたライティング アシスタントです。ユーザーのリクエストにできる限り応えられる最高のブログ記事を以下の推奨事項留意点に注意しながら具体的な例による深みとリアリティを交えながら作成しようとします。ユーザーから批判があった場合は、以前の試みを修正したバージョンで応答します。 推奨事項留意点 ・具体的な事例を加える: 各トピックについての具体的な事例や成功事例を加えることで、文章の深さと説得力を高めることができます。 ・用語の繰り返しを避ける: 代名詞や関連する用語を使用することで、用語の繰り返しを避けることができます。 ・抽象的な表現を具体的にする: 抽象的な用語についての具体的な説明を加えることで、読者がより深く理解できるでしょう。 ・段落の構成を明確化する: 各段落の目的と内容を明確化することで、文章の全体的な流れと構成を改善することができます。 ''', llm_config={"config_list": config_list, "cache_seed": None}, ) reflection_assistant = AssistantAgent( name="reflection_assistant", system_message="執筆内容に関する批評と推奨事項を作成します。具体的な例による深みとリアリティを交えて、長さ、深さ、スタイルなどのリクエストを含む詳細な推奨事項を提供します。", llm_config={"config_list": config_list, "cache_seed": None}, ) '''@user_proxy.register_for_execution() @writing_assistant.register_for_llm(description="部門の売上を取り出す関数") def extract_cagr(division: Annotated[str, "部門名（営業部、カスタマーサポート部、製品開発部、マーケティング部、オンライン販売部）"]) -> str: select_one_data_query = "SELECT `部門`,FORMAT(SUM(`売上高`),0) as `売上` FROM corp_jkpi group by `部門`" cursor.execute(select_one_data_query) result = cursor.fetchall() return result''' def reflection_message(recipient, messages, sender, config): print("Reflecting...") return f"次の文章について考察し、批評してください。 \n\n {recipient.chat_messages_for_summary(sender)[-1]['content']}" nested_chat_queue = [ { "recipient": reflection_assistant, "message": reflection_message, "max_turns": 1, }, ] user_proxy.register_nested_chats( nested_chat_queue, trigger=writing_assistant, ) def generate_blog_article(prompt): #global token #token = "" #token = get_next_token() global config_list config_list.clear config_list = [ {"model": "llama-3.3-70b-versatile", "api_key": token, "api_type": "groq"}, {"model": "openai/gpt-oss-120b", "api_key": token, "api_type": "groq"}, ] #print("config token: ",token) llmresp = get_llm(prompt) with Cache.disk(cache_seed=42) as cache: resp = user_proxy.initiate_chat( writing_assistant, message=prompt, max_turns=2, cache=cache, ) content_list = [] for i in range(len(resp.chat_history)): content_list.append(resp.chat_history[i]["content"]) compared = compare_result("[eprag]\n\n".join(content_list),"[llm]"+llmresp) return "\n\n".join(content_list),llmresp,compared # Gradioインターフェースの作成 default_prompt = ( '"経営コンサルティング業界の最新アップデートに関する魅力的なブログ記事を書いてください。"\n' '"ブログ記事は、一般の読者にとって魅力的で理解しやすいものでなければなりません。"\n' '"段落数は 3 段落以上、1,000 語以内である必要があります。"' ) myinput = gr.Textbox(lines=10, placeholder="ここにプロンプトを入力してください", value=default_prompt, label="プロンプト") myoutput = gr.Textbox(lines=10, placeholder="生成されたブログ記事", label="生成されたブログ記事") mygptout = gr.Textbox(lines=10, placeholder="生成されたGPTブログ記事", label="生成されたGPTブログ記事") myhtml = gr.Textbox(lines=10, placeholder="生成された比較分析", label="生成された比較分析") with gr.Blocks(title="ブログ記事生成（LLM Reflextion）") as demo: gr.Markdown("# ブログ記事生成（LLM Reflextion）") gr.Markdown("経営コンサルティング業界の最新アップデートに関する魅力的なブログ記事を書いてください。") with gr.Row(): myinput.render() with gr.Row(): myoutput.render() with gr.Row(): mygptout.render() with gr.Row(): submit_btn = gr.Button("生成") clear_btn = gr.Button("クリア") submit_btn.click(generate_blog_article, inputs=myinput, outputs=[myoutput,mygptout,myhtml]) clear_btn.click(lambda: ("", "", ""), None, [myoutput, mygptout, myhtml]) #gr.Markdown("Reflextionコメント") with gr.Row(visible=False) as html_row: myhtml.render() submit_btn.click(lambda: gr.update(visible=True), None, html_row) demo.launch()