From 6aea19f4dfc914aa706574088f819e8f70a25570 Mon Sep 17 00:00:00 2001 From: hamakyo Date: Sun, 12 Apr 2026 01:47:03 +0900 Subject: [PATCH 1/2] Improve CLI reliability and tooling --- .env | 8 - .env.example | 6 + .github/workflows/ci.yml | 38 ++ .gitignore | 4 + README.md | 93 +++++ mogisystem.py | 872 +++++++++++++++++++++++++++------------ pyproject.toml | 16 + requirements-dev.txt | 3 + requirements.txt | 9 +- tests/test_mogisystem.py | 380 +++++++++++++++++ 10 files changed, 1153 insertions(+), 276 deletions(-) delete mode 100644 .env create mode 100644 .env.example create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt create mode 100644 tests/test_mogisystem.py diff --git a/.env b/.env deleted file mode 100644 index 8aa86b1..0000000 --- a/.env +++ /dev/null @@ -1,8 +0,0 @@ -OPENAI_API_KEY= -OPENAI_MODEL=gpt-4o-mini - -ANTHROPIC_API_KEY= -ANTHROPIC_MODEL=claude-3-5-haiku-20241022 - -GOOGLE_API_KEY=<> -GEMINI_MODEL=gemini-1.5-flash \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..318ec95 --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +OPENAI_API_KEY=your-openai-api-key +OPENAI_MODEL=gpt-4 +ANTHROPIC_API_KEY=your-anthropic-api-key +ANTHROPIC_MODEL=claude-3-opus-20240229 +GOOGLE_API_KEY=your-google-api-key +GEMINI_MODEL=gemini-pro diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..65ba925 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,38 @@ +name: CI + +on: + push: + branches: + - main + - master + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + timeout-minutes: 10 + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: "pip" + + - name: Install dependencies + run: pip install -r requirements-dev.txt + + - name: Lint + run: ruff check . + + - name: Type check + run: mypy + + - name: Validate syntax + run: python -m py_compile mogisystem.py tests/test_mogisystem.py + + - name: Run tests + run: python -m unittest discover -s tests -v diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..762b8d7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.env +.venv/ +__pycache__/ +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..ca740db --- /dev/null +++ b/README.md @@ -0,0 +1,93 @@ +# MOGI System + +複数のLLMに順番に議論させ、最後に結論を生成するシンプルなCLIです。 + +## セットアップ + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -r requirements-dev.txt +cp .env.example .env +``` + +前提 Python は `3.11.x` です。 + +`.env` に使いたいプロバイダのAPIキーを設定してください。未設定のプロバイダはスキップされ、設定済みのプロバイダだけで実行されます。 + +## 使い方 + +対話入力: + +```bash +python mogisystem.py +``` + +引数指定: + +```bash +python mogisystem.py "生成AIの教育利用" --rounds 2 --sleep-seconds 0 +``` + +quiet 実行: + +```bash +python mogisystem.py "生成AIの教育利用" --quiet +``` + +`--quiet` は text 出力時に ASCII アートとラウンド進捗を抑止し、最終結果だけを表示します。 + +provider 指定: + +```bash +python mogisystem.py "生成AIの教育利用" --providers gpt,claude --rounds 2 --timeout-seconds 30 --retry-count 1 +``` + +`--providers` を付けた場合は、指定した provider が利用できないとエラーで終了します。 +`--timeout-seconds` で各 provider 呼び出しの上限時間を指定できます。`0` 以下なら無効化します。 +`--retry-count` で timeout や一時的な API エラー時の再試行回数を指定できます。 + +JSON 出力: + +```bash +python mogisystem.py "生成AIの教育利用" --providers gpt,claude --rounds 2 --sleep-seconds 0 --output json +``` + +Markdown 出力: + +```bash +python mogisystem.py "生成AIの教育利用" --providers gpt,claude --rounds 2 --output markdown +``` + +`--output json` と `--output markdown` を使う場合は `topic` を引数で渡してください。 + +ファイル保存: + +```bash +python mogisystem.py "生成AIの教育利用" --output markdown --output-file outputs/result.md +``` + +`--output-file` を使うと、標準出力と同じ内容を指定ファイルにも保存します。親ディレクトリがなければ自動で作成します。 + +ヘルプ: + +```bash +python mogisystem.py --help +``` + +## テスト + +```bash +python -m unittest discover -s tests -v +``` + +## 静的解析 + +```bash +ruff check . +mypy +``` + +## セキュリティ + +このリポジトリでは `.env` を Git 管理しない前提です。既に `.env` が追跡対象になっている場合は、履歴や共有先にAPIキーが残っていないか確認し、必要ならキーをローテーションしてください。 diff --git a/mogisystem.py b/mogisystem.py index 37fea02..5c7b88f 100644 --- a/mogisystem.py +++ b/mogisystem.py @@ -1,263 +1,609 @@ -import os -import warnings -from typing import List, Dict, Optional -import anthropic -import google.generativeai as genai -from openai import OpenAI -import time -import dotenv -from dataclasses import dataclass -from enum import Enum - -@dataclass -class AIResponse: - content: str - error: Optional[str] = None - -class AIModel(Enum): - GPT = "gpt" - CLAUDE = "claude" - GEMINI = "gemini" - -class MAGISystem: - def __init__(self): - dotenv.load_dotenv() - self._initialize_clients() - self._initialize_roles() - - def _initialize_clients(self) -> None: - """APIクライアントの初期化""" - try: - # OpenAI - self.openai_client = OpenAI(api_key=self._get_env_var("OPENAI_API_KEY")) - self.openai_model = self._get_env_var("OPENAI_MODEL", default="gpt-4") - - # Anthropic - self.anthropic_client = anthropic.Anthropic(api_key=self._get_env_var("ANTHROPIC_API_KEY")) - self.anthropic_model = self._get_env_var("ANTHROPIC_MODEL", default="claude-3-opus-20240229") - - # Google - genai.configure(api_key=self._get_env_var("GOOGLE_API_KEY")) - self.gemini_model = self._get_env_var("GEMINI_MODEL", default="gemini-pro") - self.gemini = genai.GenerativeModel(self.gemini_model) - except Exception as e: - raise RuntimeError(f"クライアントの初期化に失敗しました: {str(e)}") - - def _get_env_var(self, key: str, default: Optional[str] = None) -> str: - """環境変数の取得と検証""" - value = os.getenv(key, default) - if value is None: - raise ValueError(f"環境変数 {key} が設定されていません") - return value - - def _initialize_roles(self) -> None: - """AI役割の定義""" - self.roles = { - AIModel.GPT: "あなたは論理的な分析と批判的思を得意とする議論者です。", - AIModel.CLAUDE: "あなたは幅広い知識と創造的な発想を持つ議論者です。", - AIModel.GEMINI: "あなたは実践的でバランスの取れた視点を持つ議論者です。" - } - - def get_ai_response(self, model: AIModel, prompt: str) -> AIResponse: - """統一されたインターフェースでAIレスポンスを取得""" - try: - if model == AIModel.GPT: - return self._get_gpt_response(prompt) - elif model == AIModel.CLAUDE: - return self._get_claude_response(prompt) - elif model == AIModel.GEMINI: - return self._get_gemini_response(prompt) - except Exception as e: - return AIResponse(content="", error=str(e)) - -def _get_gpt_response(self, prompt: str) -> AIResponse: - """GPT-4からの応答を取得""" - response = self.openai_client.chat.completions.create( - model=self.openai_model, - messages=[ - {"role": "system", "content": self.roles[AIModel.GPT]}, - {"role": "user", "content": prompt} - ], - stream=False # Streamlitでは非ストリーミングモードを使用 - ) - - content = response.choices[0].message.content - return AIResponse(content=content) - -def _get_claude_response(self, prompt: str) -> AIResponse: - """Claudeからの応答を取得""" - response = self.anthropic_client.messages.create( - model=self.anthropic_model, - max_tokens=1000, - system=self.roles[AIModel.CLAUDE], - messages=[{"role": "user", "content": prompt}], - stream=False # Streamlitでは非ストリーミングモードを使用 - ) - - content = response.content[0].text - return AIResponse(content=content) - -def _get_gemini_response(self, prompt: str) -> AIResponse: - """Geminiからの応答を取得""" - try: - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - response = self.gemini.generate_content( - f"{self.roles[AIModel.GEMINI]}\n\n{prompt}", - safety_settings={"HARM_CATEGORY_HARASSMENT": "BLOCK_NONE"}, - generation_config={ - "temperature": 0.7, - "top_p": 0.8, - "top_k": 40 - }, - stream=False # Streamlitでは非ストリーミングモードを使用 - ) - - content = response.text - return AIResponse(content=content) - - except Exception as e: - return AIResponse(content="", error=f"Geminiでエラーが発生: {str(e)}") - - def _generate_conclusion(self, discussion_history: List[Dict]) -> Dict: - """最終結論を生成""" - try: - conclusion_prompt = f""" - これまでの議論全体を踏まえて、最終的な結論を導き出してください。 - 以下が議論の履歴です：{discussion_history} - """ - conclusion = self.get_ai_response(AIModel.CLAUDE, conclusion_prompt) - return { - "discussion_history": discussion_history, - "final_conclusion": conclusion.content - } - except Exception as e: - print(f"\n### ⚠️ 結論の導出時にエラーが発生しました: {str(e)}") - return { - "discussion_history": discussion_history, - "final_conclusion": "結論の導出に失敗しました。" - } - - def facilitate_discussion(self, topic: str, rounds: int = 3) -> Dict: - """AIによる議論を進行""" - discussion_history = [] - current_prompt = topic - - for round_num in range(rounds): - print(f"\n## 🤖 ラウンド {round_num + 1}") - - try: - responses = self._get_round_responses(current_prompt) - discussion_history.append({ - "round": round_num + 1, - **responses - }) - current_prompt = self._prepare_next_prompt(responses) - print("\n---") - time.sleep(2) - - except Exception as e: - print(f"\n### ⚠️ エラーが発生しました: {str(e)}") - continue - - # 最終結論の生成 - conclusion_prompt = f""" - これまでの議論を総括し、以下の点を含めた独自の結論を導き出してください： - 1. これまでの議論で見落とされていた視点 - 2. 各AIの意見の中で特に重要だと思われるポイント - 3. 今後の展望や提言 - - ※これまでの発言を単に要約するのではなく、新しい視点で分析してください。 - ※以下が議論の履歴です： - {discussion_history} - """ - - try: - conclusion = self.get_ai_response(AIModel.CLAUDE, conclusion_prompt) - return { - "discussion_history": discussion_history, - "final_conclusion": conclusion.content - } - except Exception as e: - return { - "discussion_history": discussion_history, - "final_conclusion": f"結論の導出に失敗しました: {str(e)}" - } - - def _get_round_responses(self, prompt: str) -> Dict[str, str]: - """1ラウンドの応答を取得""" - responses = {} - - # GPTの応答 - gpt_response = self.get_ai_response(AIModel.GPT, prompt) - responses["gpt"] = gpt_response.content - - # Claudeの応答 - claude_prompt = f"{prompt}\n\nGPT-4の意見: {gpt_response.content}" - claude_response = self.get_ai_response(AIModel.CLAUDE, claude_prompt) - responses["claude"] = claude_response.content - - # Geminiの応答 - gemini_prompt = f"{claude_prompt}\nClaudeの意見: {claude_response.content}" - gemini_response = self.get_ai_response(AIModel.GEMINI, gemini_prompt) - responses["gemini"] = gemini_response.content - - return responses - - def _prepare_next_prompt(self, responses: Dict[str, str]) -> str: - """次のラウンドのプロンプトを生成""" - return f""" - これまでの議論を踏まえて、さらなる検討点や新しい視点を提示してください。 - - 議論の経緯: - GPT: {responses["gpt"]} - Claude: {responses["claude"]} - Gemini: {responses["gemini"]} - """ - -def display_ascii_art(): - """ASCIIアートを表示""" - ascii_art = """ - ███╗ ███╗ ██████╗ ██████╗ ██╗ ███████╗██╗ ██╗███████╗ - ████╗ ████║██╔══██╗██╔════╝ ██║ ██╔════╝╚██╗ ██╔╝██╔════╝ - ██╔████╔██║██║ ██║██║ ███╗██║█████╗███████╗ ╚████╔╝ ███████╗ - ██║╚██╔╝██║██║ ██║██║ ██║██║╚════╝╚════██║ ╚██╔╝ ╚════██║ - ██║ ╚═╝ ██║╚██████╔╝╚██████╔╝██║ ███████║ ██║ ███████║ - ╚═╝ ╚═╝ ╚═════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═╝ ╚══════╝ - ============= MOGI-SYSTEM: Multi-AI Generative Intelligence System ============= - """ - print(ascii_art) - -def main(): - """メイン実行関数""" - try: - display_ascii_art() - magi = MAGISystem() - - print("\n💭 議論のテーマを入力してください：") - user_topic = input().strip() - - topic = f""" - テーマ：「{user_topic}」 - - このテーマについて、異なる視点から議論を展開し、 - 具体的な例を挙げながら検討してください。 - """ - - # 議論を開始 - result = magi.facilitate_discussion(topic) - - # 結果の出力 - print("\n=== 最終結論 ===") - print(result["final_conclusion"]) - - except Exception as e: - print(f"\n⚠️ エラーが発生しました: {str(e)}") - print("プログラムを終了します。") - return 1 - - return 0 - -if __name__ == "__main__": - exit_code = main() - exit(exit_code) \ No newline at end of file +import argparse +import json +import os +import sys +import threading +import time +import warnings +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from typing import Any, Callable, Optional + +import dotenv + + +@dataclass +class AIResponse: + content: str + error: Optional[str] = None + + +class AIModel(Enum): + GPT = "gpt" + CLAUDE = "claude" + GEMINI = "gemini" + + @classmethod + def from_str(cls, value: str) -> "AIModel": + normalized = value.strip().lower() + for model in cls: + if model.value == normalized: + return model + valid_models = ", ".join(model.value for model in cls) + raise ValueError(f"無効な provider です: {value}. 利用可能: {valid_models}") + + +class MAGISystem: + def __init__( + self, + *, + load_env: bool = True, + sleep_seconds: float = 2.0, + timeout_seconds: float = 60.0, + retry_count: int = 0, + requested_models: list[AIModel] | None = None, + ): + if load_env: + dotenv.load_dotenv() + + self.sleep_seconds = max(0.0, sleep_seconds) + self.timeout_seconds = timeout_seconds + self.retry_count = max(0, retry_count) + self.requested_models = requested_models or list(AIModel) + self.strict_model_selection = requested_models is not None + self.available_models: list[AIModel] = [] + self.client_errors: dict[AIModel, str] = {} + self.openai_client: Any | None = None + self.anthropic_client: Any | None = None + self.gemini: Any | None = None + self._initialize_roles() + self._initialize_clients() + + def _initialize_roles(self) -> None: + """AI役割の定義""" + self.roles: dict[AIModel, str] = { + AIModel.GPT: "あなたは論理的な分析と批判的思考を得意とする議論者です。", + AIModel.CLAUDE: "あなたは幅広い知識と創造的な発想を持つ議論者です。", + AIModel.GEMINI: "あなたは実践的でバランスの取れた視点を持つ議論者です。", + } + + def _initialize_clients(self) -> None: + """利用可能なAPIクライアントだけを初期化する""" + self.openai_model = os.getenv("OPENAI_MODEL", "gpt-4") + self.anthropic_model = os.getenv("ANTHROPIC_MODEL", "claude-3-opus-20240229") + self.gemini_model = os.getenv("GEMINI_MODEL", "gemini-pro") + + self._initialize_openai_client() + self._initialize_anthropic_client() + self._initialize_gemini_client() + + if self.strict_model_selection: + missing_models = [ + model for model in self.requested_models if model not in self.available_models + ] + if missing_models: + details = "; ".join( + f"{model.value}: {self.client_errors.get(model, '利用できません')}" + for model in missing_models + ) + raise RuntimeError(f"指定された provider を利用できません: {details}") + + if not self.available_models: + details = "; ".join(self.client_errors.values()) or "有効なプロバイダがありません。" + raise RuntimeError(f"クライアントの初期化に失敗しました: {details}") + + def _initialize_openai_client(self) -> None: + if AIModel.GPT not in self.requested_models: + return + + api_key = os.getenv("OPENAI_API_KEY") + if not api_key: + self.client_errors[AIModel.GPT] = "OPENAI_API_KEY が設定されていません" + return + + try: + from openai import OpenAI + + self.openai_client = OpenAI(api_key=api_key) + self.available_models.append(AIModel.GPT) + except Exception as exc: + self.client_errors[AIModel.GPT] = f"OpenAI クライアントの初期化に失敗しました: {exc}" + + def _initialize_anthropic_client(self) -> None: + if AIModel.CLAUDE not in self.requested_models: + return + + api_key = os.getenv("ANTHROPIC_API_KEY") + if not api_key: + self.client_errors[AIModel.CLAUDE] = "ANTHROPIC_API_KEY が設定されていません" + return + + try: + import anthropic + + self.anthropic_client = anthropic.Anthropic(api_key=api_key) + self.available_models.append(AIModel.CLAUDE) + except Exception as exc: + self.client_errors[AIModel.CLAUDE] = ( + f"Anthropic クライアントの初期化に失敗しました: {exc}" + ) + + def _initialize_gemini_client(self) -> None: + if AIModel.GEMINI not in self.requested_models: + return + + api_key = os.getenv("GOOGLE_API_KEY") + if not api_key: + self.client_errors[AIModel.GEMINI] = "GOOGLE_API_KEY が設定されていません" + return + + try: + import google.generativeai as genai + + genai.configure(api_key=api_key) + self.gemini = genai.GenerativeModel(self.gemini_model) + self.available_models.append(AIModel.GEMINI) + except Exception as exc: + self.client_errors[AIModel.GEMINI] = f"Gemini クライアントの初期化に失敗しました: {exc}" + + def is_model_available(self, model: AIModel) -> bool: + return model in self.available_models + + def get_ai_response(self, model: AIModel, prompt: str) -> AIResponse: + """統一されたインターフェースでAIレスポンスを取得""" + if not self.is_model_available(model): + error = self.client_errors.get(model, f"{model.value} は利用できません") + return AIResponse(content="", error=error) + + handler = self._get_model_handler(model) + if handler is None: + return AIResponse(content="", error=f"未対応のモデルです: {model.value}") + + last_response = AIResponse(content="", error=f"{model.value} returned no response") + for attempt in range(self.retry_count + 1): + try: + response = self._run_with_timeout(model, handler, prompt) + except Exception as exc: + response = AIResponse(content="", error=str(exc)) + + if not response.error: + return response + + last_response = response + if attempt < self.retry_count: + continue + + return last_response + + def _get_model_handler(self, model: AIModel) -> Callable[[str], AIResponse] | None: + handlers: dict[AIModel, Callable[[str], AIResponse]] = { + AIModel.GPT: self._get_gpt_response, + AIModel.CLAUDE: self._get_claude_response, + AIModel.GEMINI: self._get_gemini_response, + } + return handlers.get(model) + + def _run_with_timeout( + self, + model: AIModel, + handler: Callable[[str], AIResponse], + prompt: str, + ) -> AIResponse: + if self.timeout_seconds <= 0: + return handler(prompt) + + response_holder: dict[str, AIResponse] = {} + error_holder: dict[str, Exception] = {} + + def target() -> None: + try: + response_holder["response"] = handler(prompt) + except Exception as exc: + error_holder["error"] = exc + + worker = threading.Thread(target=target, daemon=True) + worker.start() + worker.join(self.timeout_seconds) + + if worker.is_alive(): + return AIResponse( + content="", + error=( + f"{model.value} request timed out after " + f"{self.timeout_seconds:g} seconds" + ), + ) + if "error" in error_holder: + raise error_holder["error"] + return response_holder.get( + "response", + AIResponse(content="", error=f"{model.value} returned no response"), + ) + + def _get_gpt_response(self, prompt: str) -> AIResponse: + if self.openai_client is None: + return AIResponse(content="", error="OpenAI client is not initialized") + + response = self.openai_client.chat.completions.create( + model=self.openai_model, + messages=[ + {"role": "system", "content": self.roles[AIModel.GPT]}, + {"role": "user", "content": prompt}, + ], + stream=False, + ) + content = response.choices[0].message.content or "" + return AIResponse(content=content) + + def _get_claude_response(self, prompt: str) -> AIResponse: + if self.anthropic_client is None: + return AIResponse(content="", error="Anthropic client is not initialized") + + response = self.anthropic_client.messages.create( + model=self.anthropic_model, + max_tokens=1000, + system=self.roles[AIModel.CLAUDE], + messages=[{"role": "user", "content": prompt}], + ) + content = response.content[0].text + return AIResponse(content=content) + + def _get_gemini_response(self, prompt: str) -> AIResponse: + if self.gemini is None: + return AIResponse(content="", error="Gemini client is not initialized") + + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + response = self.gemini.generate_content( + f"{self.roles[AIModel.GEMINI]}\n\n{prompt}", + generation_config={ + "temperature": 0.7, + "top_p": 0.8, + "top_k": 40, + }, + ) + content = response.text + return AIResponse(content=content) + except Exception as exc: + return AIResponse(content="", error=f"Geminiでエラーが発生しました: {exc}") + + def _response_or_placeholder(self, model: AIModel, prompt: str) -> str: + response = self.get_ai_response(model, prompt) + if response.error: + return f"[{model.value} unavailable: {response.error}]" + return response.content + + def _serialize_history(self, discussion_history: list[dict[str, Any]]) -> str: + return json.dumps(discussion_history, ensure_ascii=False, indent=2) + + def _select_conclusion_model(self) -> AIModel: + if self.is_model_available(AIModel.CLAUDE): + return AIModel.CLAUDE + return self.available_models[-1] + + def _generate_conclusion(self, discussion_history: list[dict[str, Any]]) -> dict[str, Any]: + """最終結論を生成""" + conclusion_prompt = f""" +これまでの議論を総括し、以下の点を含めた独自の結論を導き出してください： +1. これまでの議論で見落とされていた視点 +2. 各AIの意見の中で特に重要だと思われるポイント +3. 今後の展望や提言 + +※これまでの発言を単に要約するのではなく、新しい視点で分析してください。 +※以下が議論の履歴です： +{self._serialize_history(discussion_history)} +""" + model = self._select_conclusion_model() + conclusion = self.get_ai_response(model, conclusion_prompt) + if conclusion.error: + return { + "discussion_history": discussion_history, + "final_conclusion": f"結論の導出に失敗しました: {conclusion.error}", + } + return { + "discussion_history": discussion_history, + "final_conclusion": conclusion.content, + } + + def facilitate_discussion( + self, + topic: str, + rounds: int = 3, + *, + show_progress: bool = True, + ) -> dict[str, Any]: + """AIによる議論を進行""" + if rounds < 1: + raise ValueError("rounds には 1 以上の値を指定してください") + + discussion_history: list[dict[str, Any]] = [] + current_prompt = topic + + for round_num in range(rounds): + if show_progress: + print(f"\n## ラウンド {round_num + 1}") + responses = self._get_round_responses(current_prompt) + discussion_history.append({"round": round_num + 1, **responses}) + current_prompt = self._prepare_next_prompt(responses) + if show_progress: + print("\n---") + if show_progress and self.sleep_seconds: + time.sleep(self.sleep_seconds) + + return self._generate_conclusion(discussion_history) + + def _get_round_responses(self, prompt: str) -> dict[str, str]: + """1ラウンド分の応答を取得する""" + responses: dict[str, str] = {} + discussion_context = prompt + + for model in self.requested_models: + response = self._response_or_placeholder(model, discussion_context) + responses[model.value] = response + discussion_context = self._build_contextual_prompt(prompt, responses) + + return responses + + def _prepare_next_prompt(self, responses: dict[str, str]) -> str: + """次のラウンドのプロンプトを生成""" + response_history = "\n".join( + f"{model.value.upper()}: {responses[model.value]}" + for model in self.requested_models + if model.value in responses + ) + return f""" +これまでの議論を踏まえて、さらなる検討点や新しい視点を提示してください。 + +議論の経緯: +{response_history} +""" + + def _build_contextual_prompt(self, base_prompt: str, responses: dict[str, str]) -> str: + if not responses: + return base_prompt + + prior_opinions = "\n".join( + f"{model.value.upper()}の意見: {responses[model.value]}" + for model in self.requested_models + if model.value in responses + ) + return f"{base_prompt}\n\nこれまでの意見:\n{prior_opinions}" + + +def display_ascii_art() -> None: + """ASCIIアートを表示""" + ascii_art = """ + ███╗ ███╗ ██████╗ ██████╗ ██╗ ███████╗██╗ ██╗███████╗ + ████╗ ████║██╔══██╗██╔════╝ ██║ ██╔════╝╚██╗ ██╔╝██╔════╝ + ██╔████╔██║██║ ██║██║ ███╗██║█████╗███████╗ ╚████╔╝ ███████╗ + ██║╚██╔╝██║██║ ██║██║ ██║██║╚════╝╚════██║ ╚██╔╝ ╚════██║ + ██║ ╚═╝ ██║╚██████╔╝╚██████╔╝██║ ███████║ ██║ ███████║ + ╚═╝ ╚═╝ ╚═════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═╝ ╚══════╝ + ============= MOGI-SYSTEM: Multi-AI Generative Intelligence System ============= + """ + print(ascii_art) + + +def build_topic_prompt(user_topic: str) -> str: + return f""" +テーマ：「{user_topic}」 + +このテーマについて、異なる視点から議論を展開し、 +具体的な例を挙げながら検討してください。 +""" + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Multi-AI discussion runner") + parser.add_argument("topic", nargs="?", help="議論のテーマ") + parser.add_argument("--rounds", type=int, default=3, help="議論ラウンド数") + parser.add_argument( + "--providers", + type=parse_providers, + help="利用する provider をカンマ区切りで指定する 例: gpt,claude", + ) + parser.add_argument( + "--output", + choices=("text", "json", "markdown"), + default="text", + help="出力形式", + ) + parser.add_argument( + "--output-file", + help="出力内容を保存するファイルパス", + ) + parser.add_argument( + "--quiet", + action="store_true", + help="text 出力時の進捗表示を抑止して最終結果だけを表示する", + ) + parser.add_argument( + "--no-ascii-art", + action="store_true", + help="起動時のASCIIアートを非表示にする", + ) + parser.add_argument( + "--sleep-seconds", + type=float, + default=2.0, + help="各ラウンド間の待機秒数", + ) + parser.add_argument( + "--timeout-seconds", + type=float, + default=60.0, + help="各 provider 呼び出しのタイムアウト秒数。0 以下で無効化", + ) + parser.add_argument( + "--retry-count", + type=int, + default=0, + help="各 provider 呼び出しの再試行回数", + ) + return parser.parse_args(argv) + + +def parse_providers(value: str) -> list[AIModel]: + providers: list[AIModel] = [] + seen: set[AIModel] = set() + + for raw_provider in value.split(","): + provider = raw_provider.strip() + if not provider: + raise argparse.ArgumentTypeError("provider に空文字は指定できません") + + try: + model = AIModel.from_str(provider) + except ValueError as exc: + raise argparse.ArgumentTypeError(str(exc)) from exc + + if model not in seen: + providers.append(model) + seen.add(model) + + if not providers: + raise argparse.ArgumentTypeError("provider を 1 つ以上指定してください") + + return providers + + +def build_output_payload( + *, + user_topic: str, + rounds: int, + available_models: list[AIModel], + requested_models: list[AIModel] | None, + result: dict[str, Any], +) -> dict[str, Any]: + return { + "topic": user_topic, + "rounds": rounds, + "requested_models": ( + [model.value for model in requested_models] if requested_models else None + ), + "available_models": [model.value for model in available_models], + **result, + } + + +def render_markdown(payload: dict[str, Any]) -> str: + discussion_sections: list[str] = [] + + for round_entry in payload["discussion_history"]: + sections = [f"### Round {round_entry['round']}"] + for speaker, content in round_entry.items(): + if speaker == "round": + continue + sections.append(f"#### {speaker.upper()}\n\n{content}") + discussion_sections.append("\n\n".join(sections)) + + requested_models = payload.get("requested_models") or ["all available"] + available_models = payload.get("available_models") or [] + + markdown_sections = [ + "# MOGI System Report", + "## Summary", + f"- Topic: {payload['topic']}", + f"- Rounds: {payload['rounds']}", + f"- Requested Models: {', '.join(requested_models)}", + f"- Available Models: {', '.join(available_models)}", + "## Discussion History", + "\n\n".join(discussion_sections) if discussion_sections else "No discussion history.", + "## Final Conclusion", + payload["final_conclusion"], + ] + return "\n\n".join(markdown_sections) + "\n" + + +def render_result(payload: dict[str, Any], *, output_format: str) -> str: + if output_format == "json": + return f"{json.dumps(payload, ensure_ascii=False, indent=2)}\n" + if output_format == "markdown": + return render_markdown(payload) + + return f"\n=== 最終結論 ===\n{payload['final_conclusion']}\n" + + +def render_error(exc: Exception, *, output_format: str) -> str: + if output_format == "json": + return f"{json.dumps({'error': str(exc)}, ensure_ascii=False, indent=2)}\n" + if output_format == "markdown": + return f"# Error\n\n{exc}\n" + + return f"\nエラーが発生しました: {exc}\nプログラムを終了します。\n" + + +def write_output(content: str, *, output_file: str | None) -> None: + if output_file is None: + return + + destination = Path(output_file) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(content, encoding="utf-8") + + +def emit_output(content: str, *, output_file: str | None) -> None: + write_output(content, output_file=output_file) + sys.stdout.write(content) + + +def main(argv: list[str] | None = None) -> int: + """メイン実行関数""" + args = parse_args(argv) + + try: + structured_output = args.output in {"json", "markdown"} + quiet_mode = args.quiet and not structured_output + if not args.no_ascii_art and not structured_output and not quiet_mode: + display_ascii_art() + + magi = MAGISystem( + sleep_seconds=args.sleep_seconds, + timeout_seconds=args.timeout_seconds, + retry_count=args.retry_count, + requested_models=args.providers, + ) + user_topic = args.topic + + if not user_topic and structured_output: + if sys.stdin.isatty(): + raise ValueError(f"{args.output} 出力では topic 引数が必要です") + user_topic = input().strip() + elif not user_topic: + print("\n議論のテーマを入力してください:") + user_topic = input().strip() + + if not user_topic: + raise ValueError("テーマが空です") + + result = magi.facilitate_discussion( + build_topic_prompt(user_topic), + rounds=args.rounds, + show_progress=not structured_output and not quiet_mode, + ) + payload = build_output_payload( + user_topic=user_topic, + rounds=args.rounds, + available_models=magi.available_models, + requested_models=args.providers, + result=result, + ) + emit_output( + render_result(payload, output_format=args.output), + output_file=args.output_file, + ) + return 0 + except Exception as exc: + error_output = render_error(exc, output_format=args.output) + try: + write_output(error_output, output_file=getattr(args, "output_file", None)) + except Exception: + pass + sys.stdout.write(error_output) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..953ce98 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[tool.ruff] +target-version = "py311" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "B"] + +[tool.mypy] +python_version = "3.11" +files = ["mogisystem.py", "tests"] +check_untyped_defs = true +disallow_untyped_defs = true +ignore_missing_imports = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..b332730 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt +mypy>=1.11,<2 +ruff>=0.6,<1 diff --git a/requirements.txt b/requirements.txt index 37d1249..8062390 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,4 @@ -openai -anthropic -google-generativeai -python-dotenv -grpcio==1.60.1 \ No newline at end of file +anthropic==0.39.0 +google-generativeai==0.8.3 +openai==1.55.0 +python-dotenv==0.21.0 diff --git a/tests/test_mogisystem.py b/tests/test_mogisystem.py new file mode 100644 index 0000000..6ec58c0 --- /dev/null +++ b/tests/test_mogisystem.py @@ -0,0 +1,380 @@ +import io +import json +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +from mogisystem import AIModel, AIResponse, MAGISystem, build_topic_prompt, main, parse_args + + +class FakeMAGISystem(MAGISystem): + def __init__( + self, + responses: dict[AIModel, list[str | Exception]], + *, + available_models: list[AIModel] | None = None, + timeout_seconds: float = 60.0, + retry_count: int = 0, + ) -> None: + self._responses = responses + self._test_available_models = available_models or [ + AIModel.GPT, + AIModel.CLAUDE, + AIModel.GEMINI, + ] + super().__init__( + load_env=False, + sleep_seconds=0, + timeout_seconds=timeout_seconds, + retry_count=retry_count, + ) + + def _initialize_clients(self) -> None: + self.available_models = list(self._test_available_models) + self.client_errors = {} + for model in AIModel: + if model not in self.available_models: + self.client_errors[model] = f"{model.value} disabled for test" + + def get_ai_response(self, model: AIModel, prompt: str) -> AIResponse: + if model not in self.available_models: + return AIResponse(content="", error=self.client_errors[model]) + + queue = self._responses.get(model, []) + if not queue: + return AIResponse(content="", error=f"no response for {model.value}") + + response = queue.pop(0) + if isinstance(response, Exception): + return AIResponse(content="", error=str(response)) + return AIResponse(content=response) + + +class MAGISystemTests(unittest.TestCase): + def test_discussion_runs_end_to_end(self) -> None: + system = FakeMAGISystem( + { + AIModel.GPT: ["gpt round 1"], + AIModel.CLAUDE: ["claude round 1", "final conclusion"], + AIModel.GEMINI: ["gemini round 1"], + } + ) + + result = system.facilitate_discussion( + build_topic_prompt("テスト"), + rounds=1, + show_progress=False, + ) + + self.assertEqual(result["discussion_history"][0]["gpt"], "gpt round 1") + self.assertEqual(result["discussion_history"][0]["claude"], "claude round 1") + self.assertEqual(result["discussion_history"][0]["gemini"], "gemini round 1") + self.assertEqual(result["final_conclusion"], "final conclusion") + + def test_missing_provider_is_reported_in_history(self) -> None: + system = FakeMAGISystem( + { + AIModel.GPT: ["gpt only", "final from gpt fallback"], + }, + available_models=[AIModel.GPT], + ) + + result = system.facilitate_discussion( + build_topic_prompt("単独実行"), + rounds=1, + show_progress=False, + ) + + self.assertIn("claude unavailable", result["discussion_history"][0]["claude"]) + self.assertIn("gemini unavailable", result["discussion_history"][0]["gemini"]) + self.assertEqual(result["final_conclusion"], "final from gpt fallback") + + def test_rounds_must_be_positive(self) -> None: + system = FakeMAGISystem({}) + + with self.assertRaises(ValueError): + system.facilitate_discussion("topic", rounds=0) + + def test_parse_args_parses_and_deduplicates_providers(self) -> None: + args = parse_args( + [ + "topic", + "--providers", + "gemini,gpt,gemini", + "--timeout-seconds", + "12.5", + "--retry-count", + "2", + "--quiet", + ] + ) + + self.assertEqual(args.providers, [AIModel.GEMINI, AIModel.GPT]) + self.assertEqual(args.timeout_seconds, 12.5) + self.assertEqual(args.retry_count, 2) + self.assertTrue(args.quiet) + + def test_provider_timeout_returns_error_response(self) -> None: + class SlowMAGISystem(MAGISystem): + def __init__(self) -> None: + super().__init__( + load_env=False, + sleep_seconds=0, + timeout_seconds=0.01, + requested_models=[AIModel.GPT], + ) + + def _initialize_clients(self) -> None: + self.available_models = [AIModel.GPT] + self.client_errors = {} + + def _get_gpt_response(self, prompt: str) -> AIResponse: + time.sleep(0.05) + return AIResponse(content=f"late: {prompt}") + + system = SlowMAGISystem() + response = system.get_ai_response(AIModel.GPT, "timeout test") + + self.assertEqual(response.content, "") + self.assertIn("timed out", response.error or "") + + def test_provider_retry_succeeds_after_transient_error(self) -> None: + class FlakyMAGISystem(MAGISystem): + def __init__(self) -> None: + self.calls = 0 + super().__init__( + load_env=False, + sleep_seconds=0, + timeout_seconds=0, + retry_count=1, + requested_models=[AIModel.GPT], + ) + + def _initialize_clients(self) -> None: + self.available_models = [AIModel.GPT] + self.client_errors = {} + + def _get_gpt_response(self, prompt: str) -> AIResponse: + self.calls += 1 + if self.calls == 1: + raise RuntimeError("temporary failure") + return AIResponse(content=f"ok: {prompt}") + + system = FlakyMAGISystem() + response = system.get_ai_response(AIModel.GPT, "retry test") + + self.assertEqual(response.content, "ok: retry test") + self.assertIsNone(response.error) + self.assertEqual(system.calls, 2) + + def test_main_outputs_json_payload(self) -> None: + calls: dict[str, object] = {} + + class StubMAGISystem: + def __init__( + self, + *, + sleep_seconds: float = 2.0, + timeout_seconds: float = 60.0, + retry_count: int = 0, + requested_models: list[AIModel] | None = None, + ) -> None: + self.sleep_seconds = sleep_seconds + self.timeout_seconds = timeout_seconds + self.retry_count = retry_count + self.available_models = requested_models or [AIModel.GPT] + calls["requested_models"] = requested_models + + def facilitate_discussion( + self, + topic: str, + rounds: int = 3, + *, + show_progress: bool = True, + ) -> dict[str, object]: + calls["topic"] = topic + calls["rounds"] = rounds + calls["show_progress"] = show_progress + return { + "discussion_history": [{"round": 1, "gemini": "gemini", "gpt": "gpt"}], + "final_conclusion": "json conclusion", + } + + stdout = io.StringIO() + with patch("mogisystem.MAGISystem", StubMAGISystem), patch("sys.stdout", stdout): + exit_code = main( + [ + "生成AIの教育利用", + "--rounds", + "2", + "--output", + "json", + "--sleep-seconds", + "0", + "--providers", + "gemini,gpt", + ] + ) + + self.assertEqual(exit_code, 0) + payload = json.loads(stdout.getvalue()) + self.assertEqual(payload["topic"], "生成AIの教育利用") + self.assertEqual(payload["rounds"], 2) + self.assertEqual(payload["requested_models"], ["gemini", "gpt"]) + self.assertEqual(payload["available_models"], ["gemini", "gpt"]) + self.assertEqual(payload["final_conclusion"], "json conclusion") + self.assertEqual(calls["requested_models"], [AIModel.GEMINI, AIModel.GPT]) + self.assertEqual(calls["show_progress"], False) + + def test_main_json_error_is_machine_readable(self) -> None: + stdout = io.StringIO() + with ( + patch("sys.stdout", stdout), + patch("sys.stdin.isatty", return_value=True), + ): + exit_code = main(["--output", "json"]) + + self.assertEqual(exit_code, 1) + payload = json.loads(stdout.getvalue()) + self.assertIn("json 出力では topic 引数が必要です", payload["error"]) + + def test_main_outputs_markdown_payload(self) -> None: + class StubMAGISystem: + def __init__( + self, + *, + sleep_seconds: float = 2.0, + timeout_seconds: float = 60.0, + retry_count: int = 0, + requested_models: list[AIModel] | None = None, + ) -> None: + self.sleep_seconds = sleep_seconds + self.timeout_seconds = timeout_seconds + self.retry_count = retry_count + self.available_models = requested_models or [AIModel.GPT] + + def facilitate_discussion( + self, + topic: str, + rounds: int = 3, + *, + show_progress: bool = True, + ) -> dict[str, object]: + return { + "discussion_history": [ + {"round": 1, "gpt": "gpt round 1", "claude": "claude round 1"} + ], + "final_conclusion": "markdown conclusion", + } + + stdout = io.StringIO() + with patch("mogisystem.MAGISystem", StubMAGISystem), patch("sys.stdout", stdout): + exit_code = main( + [ + "Markdown テスト", + "--output", + "markdown", + "--providers", + "gpt", + ] + ) + + self.assertEqual(exit_code, 0) + content = stdout.getvalue() + self.assertIn("# MOGI System Report", content) + self.assertIn("- Topic: Markdown テスト", content) + self.assertIn("#### GPT", content) + self.assertIn("## Final Conclusion", content) + self.assertIn("markdown conclusion", content) + + def test_main_quiet_text_output_suppresses_progress(self) -> None: + class StubMAGISystem: + def __init__( + self, + *, + sleep_seconds: float = 2.0, + timeout_seconds: float = 60.0, + retry_count: int = 0, + requested_models: list[AIModel] | None = None, + ) -> None: + self.sleep_seconds = sleep_seconds + self.timeout_seconds = timeout_seconds + self.retry_count = retry_count + self.available_models = requested_models or [AIModel.GPT] + + def facilitate_discussion( + self, + topic: str, + rounds: int = 3, + *, + show_progress: bool = True, + ) -> dict[str, object]: + return { + "discussion_history": [{"round": 1, "gpt": "gpt round 1"}], + "final_conclusion": "quiet conclusion", + } + + stdout = io.StringIO() + with patch("mogisystem.MAGISystem", StubMAGISystem), patch("sys.stdout", stdout): + exit_code = main(["Quiet テスト", "--quiet"]) + + self.assertEqual(exit_code, 0) + content = stdout.getvalue() + self.assertIn("=== 最終結論 ===", content) + self.assertIn("quiet conclusion", content) + self.assertNotIn("MOGI-SYSTEM", content) + self.assertNotIn("## ラウンド", content) + + def test_main_writes_output_file(self) -> None: + class StubMAGISystem: + def __init__( + self, + *, + sleep_seconds: float = 2.0, + timeout_seconds: float = 60.0, + retry_count: int = 0, + requested_models: list[AIModel] | None = None, + ) -> None: + self.sleep_seconds = sleep_seconds + self.timeout_seconds = timeout_seconds + self.retry_count = retry_count + self.available_models = requested_models or [AIModel.GPT] + + def facilitate_discussion( + self, + topic: str, + rounds: int = 3, + *, + show_progress: bool = True, + ) -> dict[str, object]: + return { + "discussion_history": [{"round": 1, "gpt": "gpt"}], + "final_conclusion": "saved conclusion", + } + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "nested" / "result.json" + stdout = io.StringIO() + with patch("mogisystem.MAGISystem", StubMAGISystem), patch("sys.stdout", stdout): + exit_code = main( + [ + "保存テスト", + "--output", + "json", + "--output-file", + str(output_path), + ] + ) + + self.assertEqual(exit_code, 0) + self.assertTrue(output_path.exists()) + self.assertEqual(output_path.read_text(encoding="utf-8"), stdout.getvalue()) + payload = json.loads(output_path.read_text(encoding="utf-8")) + self.assertEqual(payload["topic"], "保存テスト") + self.assertEqual(payload["final_conclusion"], "saved conclusion") + + +if __name__ == "__main__": + unittest.main() From c6155b5c66cf19e64a66af2b2fac6c9cd2bed6ca Mon Sep 17 00:00:00 2001 From: hamakyo Date: Sun, 12 Apr 2026 02:01:24 +0900 Subject: [PATCH 2/2] Adjust ascii art and default models --- .env.example | 6 +++--- mogisystem.py | 34 +++++++++++++++++++--------------- 2 files changed, 22 insertions(+), 18 deletions(-) diff --git a/.env.example b/.env.example index 318ec95..8a8ba43 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,6 @@ OPENAI_API_KEY=your-openai-api-key -OPENAI_MODEL=gpt-4 +OPENAI_MODEL=gpt-4o-mini ANTHROPIC_API_KEY=your-anthropic-api-key -ANTHROPIC_MODEL=claude-3-opus-20240229 +ANTHROPIC_MODEL=claude-3.5-haiku GOOGLE_API_KEY=your-google-api-key -GEMINI_MODEL=gemini-pro +GEMINI_MODEL=gemini-2.5-flash diff --git a/mogisystem.py b/mogisystem.py index 5c7b88f..02190b3 100644 --- a/mogisystem.py +++ b/mogisystem.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from enum import Enum from pathlib import Path +from textwrap import dedent from typing import Any, Callable, Optional import dotenv @@ -374,15 +375,17 @@ def _build_contextual_prompt(self, base_prompt: str, responses: dict[str, str]) def display_ascii_art() -> None: """ASCIIアートを表示""" - ascii_art = """ - ███╗ ███╗ ██████╗ ██████╗ ██╗ ███████╗██╗ ██╗███████╗ - ████╗ ████║██╔══██╗██╔════╝ ██║ ██╔════╝╚██╗ ██╔╝██╔════╝ - ██╔████╔██║██║ ██║██║ ███╗██║█████╗███████╗ ╚████╔╝ ███████╗ - ██║╚██╔╝██║██║ ██║██║ ██║██║╚════╝╚════██║ ╚██╔╝ ╚════██║ - ██║ ╚═╝ ██║╚██████╔╝╚██████╔╝██║ ███████║ ██║ ███████║ - ╚═╝ ╚═╝ ╚═════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═╝ ╚══════╝ - ============= MOGI-SYSTEM: Multi-AI Generative Intelligence System ============= - """ + ascii_art = dedent( + """ + __ __ ___ ____ ___ ____ __ __ ___ _____ _____ __ __ + | \\/ | / _ \\ / ___| |_ _| / ___| \\ \\ / /|_ _||_ _|| ____|| \\/ | + | |\\/| || | | || | _ | | _____ \\___ \\ \\ V / | | | | | _| | |\\/| | + | | | || |_| || |_| | | | |_____| ___) | | | | | | | | |___ | | | | + |_| |_| \\___/ \\____| |___| |____/ |_| |___| |_| |_____||_| |_| + + Multi-AI Generative Intelligence System + """ + ).strip() print(ascii_art) @@ -559,12 +562,6 @@ def main(argv: list[str] | None = None) -> int: if not args.no_ascii_art and not structured_output and not quiet_mode: display_ascii_art() - magi = MAGISystem( - sleep_seconds=args.sleep_seconds, - timeout_seconds=args.timeout_seconds, - retry_count=args.retry_count, - requested_models=args.providers, - ) user_topic = args.topic if not user_topic and structured_output: @@ -578,6 +575,13 @@ def main(argv: list[str] | None = None) -> int: if not user_topic: raise ValueError("テーマが空です") + magi = MAGISystem( + sleep_seconds=args.sleep_seconds, + timeout_seconds=args.timeout_seconds, + retry_count=args.retry_count, + requested_models=args.providers, + ) + result = magi.facilitate_discussion( build_topic_prompt(user_topic), rounds=args.rounds,