Merge branch 'develop' into copy_and_paste
This commit is contained in:
@@ -1,7 +0,0 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
@@ -1,7 +0,0 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
@@ -1,7 +0,0 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
@@ -1,7 +0,0 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
@@ -4,14 +4,14 @@ from langchain_google_genai import ChatGoogleGenerativeAI
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadPromptConfig
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
print(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang
|
||||
from translation_utils import loadPromptConfig
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
|
||||
logger = logging.getLogger("langchain_google_genai")
|
||||
logger.setLevel(logging.ERROR)
|
||||
@@ -57,9 +57,19 @@ class GeminiClient:
|
||||
self.model = None
|
||||
|
||||
# プロンプト設定をYAMLファイルから読み込む
|
||||
prompt_config = loadPromptConfig(root_path, "translation_gemini.yml")
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_gemini.yml")
|
||||
self.supported_languages = list(translation_lang["Gemini_API"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.gemini_llm = None
|
||||
|
||||
@@ -91,6 +101,16 @@ class GeminiClient:
|
||||
api_key=self.api_key,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
@@ -98,6 +118,41 @@ class GeminiClient:
|
||||
output_lang=output_lang
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text}
|
||||
|
||||
201
src-python/models/translation/translation_groq.py
Normal file
201
src-python/models/translation/translation_groq.py
Normal file
@@ -0,0 +1,201 @@
|
||||
from openai import OpenAI
|
||||
from langchain_openai import ChatOpenAI
|
||||
from pydantic import SecretStr
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
def _authentication_check(api_key: str) -> bool:
|
||||
"""Check if the provided API key is valid by attempting to list models.
|
||||
"""
|
||||
try:
|
||||
client = OpenAI(
|
||||
api_key=api_key,
|
||||
base_url="https://api.groq.com/openai/v1",
|
||||
)
|
||||
client.models.list()
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _get_available_text_models(api_key: str) -> list[str]:
|
||||
"""Extract only Groq models suitable for translation and chat applications.
|
||||
"""
|
||||
client = OpenAI(
|
||||
api_key=api_key,
|
||||
base_url="https://api.groq.com/openai/v1",
|
||||
)
|
||||
res = client.models.list()
|
||||
allowed_models = []
|
||||
|
||||
for model in res.data:
|
||||
model_id = model.id
|
||||
|
||||
# 除外対象のキーワード
|
||||
exclude_keywords = [
|
||||
"whisper", # 音声認識
|
||||
"embedding", # 埋め込み
|
||||
"image", # 画像生成
|
||||
"tts", # 音声合成
|
||||
"audio", # 音声系
|
||||
"search", # 検索補助モデル
|
||||
"transcribe", # 音声→文字起こし
|
||||
"diarize", # 話者分離
|
||||
"vision" # 画像入力系
|
||||
]
|
||||
|
||||
# 除外キーワードが含まれているモデルをスキップ
|
||||
if any(kw in model_id.lower() for kw in exclude_keywords):
|
||||
continue
|
||||
|
||||
# テキスト処理用モデルのみ対象
|
||||
allowed_models.append(model_id)
|
||||
|
||||
allowed_models.sort()
|
||||
return allowed_models
|
||||
|
||||
class GroqClient:
|
||||
"""Groq API Translation wrapper using OpenAI-compatible endpoint.
|
||||
|
||||
Groq provides a fast LLM inference platform with an OpenAI-compatible API.
|
||||
The API endpoint: https://api.groq.com/openai/v1
|
||||
"""
|
||||
def __init__(self, root_path: str = None):
|
||||
self.api_key = None
|
||||
self.model = None
|
||||
self.base_url = "https://api.groq.com/openai/v1"
|
||||
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_groq.yml")
|
||||
self.supported_languages = list(translation_lang["Groq_API"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.groq_llm = None
|
||||
|
||||
def getModelList(self) -> list[str]:
|
||||
return _get_available_text_models(self.api_key) if self.api_key else []
|
||||
|
||||
def getAuthKey(self) -> str:
|
||||
return self.api_key
|
||||
|
||||
def setAuthKey(self, api_key: str) -> bool:
|
||||
result = _authentication_check(api_key)
|
||||
if result:
|
||||
self.api_key = api_key
|
||||
return result
|
||||
|
||||
def getModel(self) -> str:
|
||||
return self.model
|
||||
|
||||
def setModel(self, model: str) -> bool:
|
||||
if model in self.getModelList():
|
||||
self.model = model
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def updateClient(self) -> None:
|
||||
self.groq_llm = ChatOpenAI(
|
||||
base_url=self.base_url,
|
||||
model=self.model,
|
||||
api_key=SecretStr(self.api_key),
|
||||
streaming=False,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang,
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
]
|
||||
|
||||
resp = self.groq_llm.invoke(messages)
|
||||
content = ""
|
||||
if isinstance(resp.content, str):
|
||||
content = resp.content
|
||||
elif isinstance(resp.content, list):
|
||||
for part in resp.content:
|
||||
if isinstance(part, str):
|
||||
content += part
|
||||
elif isinstance(part, dict) and "content" in part and isinstance(part["content"], str):
|
||||
content += part["content"]
|
||||
return content.strip()
|
||||
|
||||
if __name__ == "__main__":
|
||||
AUTH_KEY = "GROQ_API_KEY"
|
||||
client = GroqClient()
|
||||
client.setAuthKey(AUTH_KEY)
|
||||
models = client.getModelList()
|
||||
if models:
|
||||
print("Available models:", models)
|
||||
model = input("Select a model: ")
|
||||
client.setModel(model)
|
||||
client.updateClient()
|
||||
print(client.translate("こんにちは世界", "Japanese", "English"))
|
||||
@@ -41,14 +41,14 @@ def _load_languages(path: str, filename: str) -> str:
|
||||
Returns:
|
||||
Absolute path to the resource file
|
||||
"""
|
||||
if os.path.exists(os.path.join(path, "_internal", "languages", "languages.yml")):
|
||||
languages_path = os.path.join(path, "_internal", "languages", "languages.yml")
|
||||
elif os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), "models", "translation", "languages", "languages.yml")):
|
||||
languages_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "models", "translation", "languages", "languages.yml")
|
||||
elif os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), "languages", "languages.yml")):
|
||||
languages_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "languages", "languages.yml")
|
||||
if os.path.exists(os.path.join(path, "_internal", "translation_settings", "languages", filename)):
|
||||
languages_path = os.path.join(path, "_internal", "translation_settings", "languages", filename)
|
||||
elif os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), "models", "translation", "translation_settings", "languages", filename)):
|
||||
languages_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "models", "translation", "translation_settings", "languages", filename)
|
||||
elif os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), "translation_settings", "languages", filename)):
|
||||
languages_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "translation_settings", "languages", filename)
|
||||
else:
|
||||
raise FileNotFoundError(f"Prompt file not found: {filename}")
|
||||
raise FileNotFoundError(f"Languages file not found: {filename}")
|
||||
with open(languages_path, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
@@ -99,7 +99,7 @@ def loadTranslationLanguages(path: str, force: bool = False) -> Dict[str, Any]:
|
||||
if _loaded and not force:
|
||||
return translation_lang
|
||||
|
||||
data = _load_languages(path, "languages/languages.yml")
|
||||
data = _load_languages(path, "languages.yml")
|
||||
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(
|
||||
|
||||
@@ -1,40 +1,42 @@
|
||||
from openai import OpenAI
|
||||
from langchain_openai import ChatOpenAI
|
||||
from pydantic import SecretStr
|
||||
import requests
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadPromptConfig
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang
|
||||
from translation_utils import loadPromptConfig
|
||||
sys.path.append(os_path.dirname(os_path.abspath(__file__)))
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
def _authentication_check(api_key: str, base_url: str | None = None) -> bool:
|
||||
def _authentication_check(base_url: str | None = None) -> bool:
|
||||
"""Check if the provided API key is valid by attempting to list models.
|
||||
"""
|
||||
try:
|
||||
client = OpenAI(api_key=api_key, base_url=base_url)
|
||||
client.models.list()
|
||||
return True
|
||||
response = requests.get(f"{base_url}/models", timeout=0.2)
|
||||
if response.status_code == 200:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _get_available_text_models(api_key: str, base_url: str | None = None) -> list[str]:
|
||||
def _get_available_text_models(base_url: str | None = None) -> list[str]:
|
||||
"""Extract the list of available text models from the LM Studio.
|
||||
"""
|
||||
try:
|
||||
client = OpenAI(api_key=api_key, base_url=base_url)
|
||||
res = client.models.list()
|
||||
models = res.data
|
||||
response = requests.get(f"{base_url}/models", timeout=0.2)
|
||||
models = response.json()["data"]
|
||||
except Exception:
|
||||
models = []
|
||||
|
||||
allowed_models = []
|
||||
for model in models:
|
||||
allowed_models.append(model.id)
|
||||
allowed_models.append(model["id"])
|
||||
|
||||
allowed_models.sort()
|
||||
return allowed_models
|
||||
@@ -48,9 +50,19 @@ class LMStudioClient:
|
||||
self.model = None
|
||||
self.base_url = base_url # None の場合は公式エンドポイント
|
||||
|
||||
prompt_config = loadPromptConfig(root_path, "translation_lmstudio.yml")
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_lmstudio.yml")
|
||||
self.supported_languages = list(translation_lang["LMStudio"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.openai_llm = None
|
||||
|
||||
@@ -58,13 +70,13 @@ class LMStudioClient:
|
||||
return self.base_url
|
||||
|
||||
def setBaseURL(self, base_url: str | None) -> None:
|
||||
result = _authentication_check(api_key=self.api_key, base_url=base_url)
|
||||
result = _authentication_check(base_url=base_url)
|
||||
if result:
|
||||
self.base_url = base_url
|
||||
return result
|
||||
|
||||
def getModelList(self) -> list[str]:
|
||||
return _get_available_text_models(api_key=self.api_key, base_url=self.base_url) if self.base_url else []
|
||||
return _get_available_text_models(base_url=self.base_url) if self.base_url else []
|
||||
|
||||
def getModel(self) -> str:
|
||||
return self.model
|
||||
@@ -84,12 +96,58 @@ class LMStudioClient:
|
||||
streaming=False,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang,
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
@@ -108,7 +166,7 @@ class LMStudioClient:
|
||||
return content.strip()
|
||||
|
||||
if __name__ == "__main__":
|
||||
client = LMStudioClient(base_url="http://192.168.68.110:1234/v1")
|
||||
client = LMStudioClient(base_url="http://127.0.0.1:1234/v1")
|
||||
models = client.getModelList()
|
||||
if models:
|
||||
print("Available models:", models)
|
||||
|
||||
@@ -3,19 +3,20 @@ from langchain_ollama import ChatOllama
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadPromptConfig
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang
|
||||
from translation_utils import loadPromptConfig
|
||||
sys.path.append(os_path.dirname(os_path.abspath(__file__)))
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
def _authentication_check(base_url: str | None = None) -> bool:
|
||||
"""Check authentication for Ollama API.
|
||||
"""
|
||||
try:
|
||||
response = requests.get(f"{base_url}")
|
||||
response = requests.get(f"{base_url}", timeout=0.2)
|
||||
if response.status_code == 200:
|
||||
return True
|
||||
else:
|
||||
@@ -47,9 +48,19 @@ class OllamaClient:
|
||||
self.model = None
|
||||
self.base_url = "http://localhost:11434"
|
||||
|
||||
prompt_config = loadPromptConfig(root_path, "translation_ollama.yml")
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_ollama.yml")
|
||||
self.supported_languages = list(translation_lang["Ollama"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.openai_llm = None
|
||||
|
||||
@@ -78,12 +89,58 @@ class OllamaClient:
|
||||
streaming=False,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang,
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
|
||||
@@ -4,13 +4,14 @@ from pydantic import SecretStr
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadPromptConfig
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang
|
||||
from translation_utils import loadPromptConfig
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
def _authentication_check(api_key: str, base_url: str | None = None) -> bool:
|
||||
"""Check if the provided API key is valid by attempting to list models.
|
||||
@@ -68,9 +69,19 @@ class OpenAIClient:
|
||||
self.model = None
|
||||
self.base_url = base_url # None の場合は公式エンドポイント
|
||||
|
||||
prompt_config = loadPromptConfig(root_path, "translation_openai.yml")
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_openai.yml")
|
||||
self.supported_languages = list(translation_lang["OpenAI_API"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.openai_llm = None
|
||||
|
||||
@@ -104,12 +115,62 @@ class OpenAIClient:
|
||||
streaming=False,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang,
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
# filter by source and take newest N
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
# format items
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
# truncate by char limit to mitigate token use
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
# assemble header and append to system prompt
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
@@ -128,7 +189,7 @@ class OpenAIClient:
|
||||
return content.strip()
|
||||
|
||||
if __name__ == "__main__":
|
||||
AUTH_KEY = "OPENAI_API_KEY"
|
||||
AUTH_KEY = input("OPENAI_API_KEY: ")
|
||||
client = OpenAIClient()
|
||||
client.setAuthKey(AUTH_KEY)
|
||||
models = client.getModelList()
|
||||
|
||||
199
src-python/models/translation/translation_openrouter.py
Normal file
199
src-python/models/translation/translation_openrouter.py
Normal file
@@ -0,0 +1,199 @@
|
||||
import requests
|
||||
from openai import OpenAI
|
||||
from langchain_openai import ChatOpenAI
|
||||
from pydantic import SecretStr
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
def _authentication_check(api_key: str) -> bool:
|
||||
"""Check if the provided API key is valid by attempting to list models.
|
||||
"""
|
||||
|
||||
url = "https://openrouter.ai/api/v1/auth/key"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {api_key}"
|
||||
}
|
||||
|
||||
r = requests.get(url, headers=headers, timeout=10)
|
||||
|
||||
return r.status_code == 200
|
||||
|
||||
def _get_available_text_models(api_key: str, base_url: str | None = None) -> list[str]:
|
||||
"""Extract only OpenRouter models suitable for translation and chat applications.
|
||||
"""
|
||||
client = OpenAI(api_key=api_key, base_url=base_url)
|
||||
res = client.models.list()
|
||||
allowed_models = []
|
||||
|
||||
for model in res.data:
|
||||
model_id = model.id
|
||||
|
||||
# 除外対象のキーワード
|
||||
exclude_keywords = [
|
||||
"whisper", # 音声認識
|
||||
"embedding", # 埋め込み
|
||||
"image", # 画像生成
|
||||
"tts", # 音声合成
|
||||
"audio", # 音声系
|
||||
"search", # 検索補助モデル
|
||||
"transcribe", # 音声→文字起こし
|
||||
"diarize", # 話者分離
|
||||
"vision" # 画像入力系
|
||||
]
|
||||
|
||||
# 除外キーワードが含まれているモデルをスキップ
|
||||
if any(kw in model_id.lower() for kw in exclude_keywords):
|
||||
continue
|
||||
|
||||
# テキスト処理用モデルのみ対象
|
||||
allowed_models.append(model_id)
|
||||
|
||||
allowed_models.sort()
|
||||
return allowed_models
|
||||
|
||||
class OpenRouterClient:
|
||||
"""OpenRouter API Translation wrapper using OpenAI-compatible endpoint.
|
||||
|
||||
OpenRouter provides access to various LLM models via a unified API.
|
||||
The API endpoint: https://openrouter.ai/api/v1
|
||||
"""
|
||||
def __init__(self, root_path: str = None):
|
||||
self.api_key = None
|
||||
self.model = None
|
||||
self.base_url = "https://openrouter.ai/api/v1"
|
||||
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_openrouter.yml")
|
||||
self.supported_languages = list(translation_lang["OpenRouter_API"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.openrouter_llm = None
|
||||
|
||||
def getModelList(self) -> list[str]:
|
||||
return _get_available_text_models(self.api_key, self.base_url) if self.api_key else []
|
||||
|
||||
def getAuthKey(self) -> str:
|
||||
return self.api_key
|
||||
|
||||
def setAuthKey(self, api_key: str) -> bool:
|
||||
result = _authentication_check(api_key)
|
||||
if result:
|
||||
self.api_key = api_key
|
||||
return result
|
||||
|
||||
def getModel(self) -> str:
|
||||
return self.model
|
||||
|
||||
def setModel(self, model: str) -> bool:
|
||||
if model in self.getModelList():
|
||||
self.model = model
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def updateClient(self) -> None:
|
||||
self.openrouter_llm = ChatOpenAI(
|
||||
base_url=self.base_url,
|
||||
model=self.model,
|
||||
api_key=SecretStr(self.api_key),
|
||||
streaming=False,
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang,
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
]
|
||||
|
||||
resp = self.openrouter_llm.invoke(messages)
|
||||
content = ""
|
||||
if isinstance(resp.content, str):
|
||||
content = resp.content
|
||||
elif isinstance(resp.content, list):
|
||||
for part in resp.content:
|
||||
if isinstance(part, str):
|
||||
content += part
|
||||
elif isinstance(part, dict) and "content" in part and isinstance(part["content"], str):
|
||||
content += part["content"]
|
||||
return content.strip()
|
||||
|
||||
if __name__ == "__main__":
|
||||
AUTH_KEY = input("OPENROUTER_API_KEY: ")
|
||||
client = OpenRouterClient()
|
||||
client.setAuthKey(AUTH_KEY)
|
||||
models = client.getModelList()
|
||||
if models:
|
||||
print("Available models:", models)
|
||||
model = input("Select a model: ")
|
||||
client.setModel(model)
|
||||
client.updateClient()
|
||||
print(client.translate("こんにちは世界", "Japanese", "English"))
|
||||
@@ -4,13 +4,14 @@ from pydantic import SecretStr
|
||||
|
||||
try:
|
||||
from .translation_languages import translation_lang
|
||||
from .translation_utils import loadPromptConfig
|
||||
from .translation_utils import loadTranslatePromptConfig
|
||||
except Exception:
|
||||
import sys
|
||||
from os import path as os_path
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
from translation_languages import translation_lang
|
||||
from translation_utils import loadPromptConfig
|
||||
from translation_languages import translation_lang, loadTranslationLanguages
|
||||
from translation_utils import loadTranslatePromptConfig
|
||||
translation_lang = loadTranslationLanguages(path=".", force=True)
|
||||
|
||||
BASE_URL = "https://api.platform.preferredai.jp/v1"
|
||||
|
||||
@@ -43,9 +44,19 @@ class PlamoClient:
|
||||
self.base_url = BASE_URL
|
||||
self.model = None
|
||||
|
||||
prompt_config = loadPromptConfig(root_path, "translation_plamo.yml")
|
||||
prompt_config = loadTranslatePromptConfig(root_path, "translation_plamo.yml")
|
||||
self.supported_languages = list(translation_lang["Plamo_API"]["source"].keys())
|
||||
self.prompt_template = prompt_config["system_prompt"]
|
||||
# history config (optional)
|
||||
self.history_cfg = prompt_config.get("history", {
|
||||
"use_history": False,
|
||||
"sources": [],
|
||||
"max_messages": 0,
|
||||
"max_chars": 0,
|
||||
"header_template": "",
|
||||
"item_template": "[{source}] {role}: {text}",
|
||||
})
|
||||
self._context_history: list[dict] = []
|
||||
|
||||
self.plamo_llm = None
|
||||
|
||||
@@ -79,12 +90,58 @@ class PlamoClient:
|
||||
api_key=SecretStr(self.api_key),
|
||||
)
|
||||
|
||||
def setContextHistory(self, history_items: list[dict]) -> None:
|
||||
"""Set recent conversation history for prompt injection.
|
||||
|
||||
Each item should be a dict containing:
|
||||
- source: "chat" | "mic" | "speaker"
|
||||
- text: message string
|
||||
- timestamp: ISO format datetime string
|
||||
"""
|
||||
self._context_history = history_items or []
|
||||
|
||||
def translate(self, text: str, input_lang: str, output_lang: str) -> str:
|
||||
system_prompt = self.prompt_template.format(
|
||||
supported_languages=self.supported_languages,
|
||||
input_lang=input_lang,
|
||||
output_lang=output_lang
|
||||
)
|
||||
|
||||
# Inject recent conversation history if enabled by YAML config
|
||||
if self.history_cfg.get("use_history"):
|
||||
allowed_sources = set(self.history_cfg.get("sources", []))
|
||||
max_messages = int(self.history_cfg.get("max_messages", 0))
|
||||
max_chars = int(self.history_cfg.get("max_chars", 0))
|
||||
item_tmpl = self.history_cfg.get("item_template", "[{source}] {role}: {text}")
|
||||
header_tmpl = self.history_cfg.get("header_template", "{history}")
|
||||
|
||||
filtered = [h for h in self._context_history if h.get("source") in allowed_sources]
|
||||
recent = filtered[-max_messages:] if max_messages > 0 else filtered
|
||||
formatted_items = []
|
||||
for h in recent:
|
||||
# Format timestamp as HH:MM to save tokens
|
||||
timestamp_str = ''
|
||||
if 'timestamp' in h:
|
||||
from datetime import datetime
|
||||
try:
|
||||
ts = datetime.fromisoformat(h['timestamp'])
|
||||
timestamp_str = ts.strftime('%H:%M')
|
||||
except:
|
||||
timestamp_str = ''
|
||||
formatted_items.append(
|
||||
item_tmpl.format(
|
||||
timestamp=timestamp_str,
|
||||
source=h.get("source", ""),
|
||||
text=h.get("text", ""),
|
||||
)
|
||||
)
|
||||
history_blob = "\n".join(formatted_items).strip()
|
||||
if max_chars and len(history_blob) > max_chars:
|
||||
history_blob = history_blob[-max_chars:]
|
||||
history_header = header_tmpl.format(max_messages=max_messages, history=history_blob)
|
||||
if history_header:
|
||||
system_prompt = f"{system_prompt}\n\n{history_header}"
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": text},
|
||||
|
||||
@@ -769,3 +769,11 @@ LMStudio:
|
||||
Ollama:
|
||||
source: *openai_langs
|
||||
target: *openai_langs
|
||||
|
||||
Groq_API:
|
||||
source: *openai_langs
|
||||
target: *openai_langs
|
||||
|
||||
OpenRouter_API:
|
||||
source: *openai_langs
|
||||
target: *openai_langs
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker] # 取り込み対象の履歴種別
|
||||
max_messages: 5 # 注入する履歴件数の上限(新しい順)
|
||||
max_chars: 4000 # 履歴整形後の最大文字数(超過時は先頭を切り捨て)
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -0,0 +1,16 @@
|
||||
system_prompt: |
|
||||
You are a helpful translation assistant.
|
||||
Supported languages:
|
||||
{supported_languages}
|
||||
|
||||
Translate the user provided text from {input_lang} to {output_lang}.
|
||||
Return ONLY the translated text. Do not add quotes or extra commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -4,4 +4,13 @@ system_prompt: |
|
||||
{supported_languages}
|
||||
|
||||
Translate the following text from {input_lang} to {output_lang}.
|
||||
output only the translated text without any additional commentary.
|
||||
output only the translated text without any additional commentary.
|
||||
history:
|
||||
use_history: true
|
||||
sources: [chat, mic, speaker]
|
||||
max_messages: 5
|
||||
max_chars: 4000
|
||||
header_template: |
|
||||
Conversation context (recent {max_messages} messages):
|
||||
{history}
|
||||
item_template: "[{timestamp}][{source}] {text}"
|
||||
@@ -15,6 +15,8 @@ try:
|
||||
from .translation_openai import OpenAIClient
|
||||
from .translation_lmstudio import LMStudioClient
|
||||
from .translation_ollama import OllamaClient
|
||||
from .translation_groq import GroqClient
|
||||
from .translation_openrouter import OpenRouterClient
|
||||
except Exception:
|
||||
import sys
|
||||
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
|
||||
@@ -25,6 +27,8 @@ except Exception:
|
||||
from translation_openai import OpenAIClient
|
||||
from translation_lmstudio import LMStudioClient
|
||||
from translation_ollama import OllamaClient
|
||||
from translation_groq import GroqClient
|
||||
from translation_openrouter import OpenRouterClient
|
||||
|
||||
import ctranslate2
|
||||
import transformers
|
||||
@@ -50,6 +54,8 @@ class Translator:
|
||||
self.plamo_client: Optional[PlamoClient] = None
|
||||
self.gemini_client: Optional[GeminiClient] = None
|
||||
self.openai_client: Optional[OpenAIClient] = None
|
||||
self.groq_client: Optional[GroqClient] = None
|
||||
self.openrouter_client: Optional[OpenRouterClient] = None
|
||||
self.lmstudio_client: LMStudioClient[LMStudioClient] = None
|
||||
self.ollama_client: OllamaClient[OllamaClient] = None
|
||||
self.ctranslate2_translator: Any = None
|
||||
@@ -176,6 +182,84 @@ class Translator:
|
||||
"""Update the OpenAI client (fetch available models)."""
|
||||
self.openai_client.updateClient()
|
||||
|
||||
def authenticationGroqAuthKey(self, auth_key: str, root_path: str = None) -> bool:
|
||||
"""Authenticate Groq API with the provided key.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
self.groq_client = GroqClient(root_path=root_path)
|
||||
if self.groq_client.setAuthKey(auth_key):
|
||||
return True
|
||||
else:
|
||||
self.groq_client = None
|
||||
return False
|
||||
|
||||
def getGroqModelList(self) -> list[str]:
|
||||
"""Get available Groq models.
|
||||
|
||||
Returns a list of model names, or an empty list on failure.
|
||||
"""
|
||||
if self.groq_client is None:
|
||||
return []
|
||||
return self.groq_client.getModelList()
|
||||
|
||||
def setGroqModel(self, model: str) -> bool:
|
||||
"""Change the Groq model used for translation.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
if self.groq_client is None:
|
||||
return False
|
||||
return self.groq_client.setModel(model)
|
||||
|
||||
def updateGroqClient(self) -> None:
|
||||
"""Update the Groq client (fetch available models)."""
|
||||
self.groq_client.updateClient()
|
||||
|
||||
def authenticationOpenRouterAuthKey(self, auth_key: str, root_path: str = None) -> bool:
|
||||
"""Authenticate OpenRouter API with the provided key.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
self.openrouter_client = OpenRouterClient(root_path=root_path)
|
||||
if self.openrouter_client.setAuthKey(auth_key):
|
||||
return True
|
||||
else:
|
||||
self.openrouter_client = None
|
||||
return False
|
||||
|
||||
def getOpenRouterModelList(self) -> list[str]:
|
||||
"""Get available OpenRouter models.
|
||||
|
||||
Returns a list of model names, or an empty list on failure.
|
||||
"""
|
||||
if self.openrouter_client is None:
|
||||
return []
|
||||
return self.openrouter_client.getModelList()
|
||||
|
||||
def setOpenRouterModel(self, model: str) -> bool:
|
||||
"""Change the OpenRouter model used for translation.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
if self.openrouter_client is None:
|
||||
return False
|
||||
return self.openrouter_client.setModel(model)
|
||||
|
||||
def updateOpenRouterClient(self) -> None:
|
||||
"""Update the OpenRouter client (fetch available models)."""
|
||||
self.openrouter_client.updateClient()
|
||||
|
||||
def getLMStudioConnected(self) -> bool:
|
||||
"""Get LM Studio connection status.
|
||||
|
||||
Returns True if connected, False otherwise.
|
||||
"""
|
||||
if self.lmstudio_client is None:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def setLMStudioClientURL(self, base_url: str | None = None, root_path: str = None) -> bool:
|
||||
"""Authenticate LM Studio with the provided base URL.
|
||||
|
||||
@@ -207,13 +291,26 @@ class Translator:
|
||||
"""Update the LM Studio client (fetch available models)."""
|
||||
self.lmstudio_client.updateClient()
|
||||
|
||||
def getOllamaConnected(self) -> bool:
|
||||
"""Get Ollama connection status.
|
||||
|
||||
Returns True if connected, False otherwise.
|
||||
"""
|
||||
if self.ollama_client is None:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def checkOllamaClient(self, root_path: str = None) -> bool:
|
||||
"""Check if Ollama client is available.
|
||||
|
||||
Returns True if Ollama is reachable, False otherwise.
|
||||
"""
|
||||
self.ollama_client = OllamaClient(root_path=root_path)
|
||||
return self.ollama_client.authenticationCheck()
|
||||
result = self.ollama_client.authenticationCheck()
|
||||
if result is False:
|
||||
self.ollama_client = None
|
||||
return result
|
||||
|
||||
def getOllamaModelList(self, root_path: str = None) -> bool:
|
||||
"""Initialize Ollama client and fetch available models.
|
||||
@@ -328,9 +425,18 @@ class Translator:
|
||||
target_language = translation_lang[translator_name]["target"][target_language]
|
||||
return source_language, target_language
|
||||
|
||||
def translate(self, translator_name: str, weight_type: str, source_language: str, target_language: str, target_country: str, message: str) -> Any:
|
||||
def translate(self, translator_name: str, weight_type: str, source_language: str, target_language: str, target_country: str, message: str, context_history: Optional[list[dict]] = None) -> Any:
|
||||
"""Translate `message` using the named translator backend.
|
||||
|
||||
Args:
|
||||
translator_name: Name of the translator backend to use
|
||||
weight_type: Model weight type for CTranslate2
|
||||
source_language: Source language name
|
||||
target_language: Target language name
|
||||
target_country: Target country for locale-specific translations
|
||||
message: Text to translate
|
||||
context_history: Optional conversation context (Chat/Mic/Speaker messages)
|
||||
|
||||
Returns translated string on success, or False on failure. When
|
||||
source_language == target_language the original message is returned.
|
||||
"""
|
||||
@@ -363,6 +469,8 @@ class Translator:
|
||||
if self.plamo_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.plamo_client.setContextHistory(context_history)
|
||||
result = self.plamo_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
@@ -372,6 +480,8 @@ class Translator:
|
||||
if self.gemini_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.gemini_client.setContextHistory(context_history)
|
||||
result = self.gemini_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
@@ -381,15 +491,41 @@ class Translator:
|
||||
if self.openai_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.openai_client.setContextHistory(context_history)
|
||||
result = self.openai_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
output_lang=target_language,
|
||||
)
|
||||
case "Groq_API":
|
||||
if self.groq_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.groq_client.setContextHistory(context_history)
|
||||
result = self.groq_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
output_lang=target_language,
|
||||
)
|
||||
case "OpenRouter_API":
|
||||
if self.openrouter_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.openrouter_client.setContextHistory(context_history)
|
||||
result = self.openrouter_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
output_lang=target_language,
|
||||
)
|
||||
case "LMStudio":
|
||||
if self.lmstudio_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.lmstudio_client.setContextHistory(context_history)
|
||||
result = self.lmstudio_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
@@ -399,6 +535,8 @@ class Translator:
|
||||
if self.ollama_client is None:
|
||||
result = False
|
||||
else:
|
||||
if context_history:
|
||||
self.ollama_client.setContextHistory(context_history)
|
||||
result = self.ollama_client.translate(
|
||||
message,
|
||||
input_lang=source_language,
|
||||
|
||||
@@ -101,16 +101,16 @@ def downloadCTranslate2Tokenizer(path: str, weight_type: str = "m2m100_418M-ct2-
|
||||
tokenizer_path = os_path.join("./weights", "ctranslate2", directory_name, "tokenizer")
|
||||
transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
|
||||
|
||||
def loadPromptConfig(root_path: str | None = None, prompt_filename: str | None = None) -> dict:
|
||||
def loadTranslatePromptConfig(root_path: str | None = None, prompt_filename: str | None = None) -> dict:
|
||||
# PyInstaller 展開後
|
||||
if root_path and prompt_filename and os_path.exists(os_path.join(root_path, "_internal", "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(root_path, "_internal", "prompt", prompt_filename)
|
||||
if root_path and prompt_filename and os_path.exists(os_path.join(root_path, "_internal", "translation_settings", "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(root_path, "_internal", "translation_settings", "prompt", prompt_filename)
|
||||
# src-python 直下実行
|
||||
elif prompt_filename and os_path.exists(os_path.join(os_path.dirname(__file__), "models", "translation", "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(os_path.dirname(__file__), "models", "translation", "prompt", prompt_filename)
|
||||
elif prompt_filename and os_path.exists(os_path.join(os_path.dirname(__file__), "models", "translation", "translation_settings", "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(os_path.dirname(__file__), "models", "translation", "translation_settings", "prompt", prompt_filename)
|
||||
# translation フォルダ直下実行
|
||||
elif prompt_filename and os_path.exists(os_path.join(os_path.dirname(__file__), "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(os_path.dirname(__file__), "prompt", prompt_filename)
|
||||
elif prompt_filename and os_path.exists(os_path.join(os_path.dirname(__file__), "translation_settings", "prompt", prompt_filename)):
|
||||
prompt_path = os_path.join(os_path.dirname(__file__), "translation_settings", "prompt", prompt_filename)
|
||||
else:
|
||||
raise FileNotFoundError(f"Prompt file not found: {prompt_filename}")
|
||||
with open(prompt_path, "r", encoding="utf-8") as f:
|
||||
|
||||
Reference in New Issue
Block a user