feat: Implement translation prompt history injection for Chat/Mic/Speaker

- Added a history management system in model.py to store and retrieve recent messages from Chat, Mic, and Speaker.
- Updated controller.py to automatically add messages to the translation history after processing.
- Enhanced translation clients (OpenAI, Gemini, Groq, etc.) to accept and utilize context history for improved translation quality.
- Introduced YAML configuration options for enabling history injection and customizing its behavior across different translation models.
- Ensured that only original messages are stored in history to optimize token usage during translation.
This commit is contained in:
misyaguziya
2025-12-15 01:15:47 +09:00
parent 9e88cff889
commit dac903e07c
18 changed files with 734 additions and 7 deletions

View File

@@ -60,6 +60,16 @@ class GeminiClient:
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}

View File

@@ -76,6 +76,16 @@ class GroqClient:
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
@@ -109,12 +119,58 @@ class GroqClient:
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},

View File

@@ -53,6 +53,16 @@ class LMStudioClient:
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
@@ -86,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},

View File

@@ -51,6 +51,16 @@ class OllamaClient:
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
@@ -79,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},

View File

@@ -72,6 +72,16 @@ class OpenAIClient:
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
@@ -105,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},

View File

@@ -76,6 +76,16 @@ class OpenRouterClient:
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
@@ -109,12 +119,58 @@ class OpenRouterClient:
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},

View File

@@ -47,6 +47,16 @@ class PlamoClient:
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
@@ -80,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},

View File

@@ -4,4 +4,13 @@ system_prompt: |
{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.
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}"

View File

@@ -5,3 +5,12 @@ system_prompt: |
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}"

View File

@@ -4,4 +4,13 @@ system_prompt: |
{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.
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}"

View File

@@ -4,4 +4,13 @@ system_prompt: |
{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.
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}"

View File

@@ -4,4 +4,13 @@ system_prompt: |
{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.
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}"

View File

@@ -5,3 +5,12 @@ system_prompt: |
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}"

View File

@@ -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}"

View File

@@ -425,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.
"""
@@ -460,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,
@@ -469,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,
@@ -478,6 +491,8 @@ 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,
@@ -487,6 +502,8 @@ class Translator:
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,
@@ -496,6 +513,8 @@ class Translator:
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,
@@ -505,6 +524,8 @@ class Translator:
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,
@@ -514,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,