Coverage for app/venv/lib/python3.14/site-packages/weblate/machinery/openai.py: 23%
140 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5from __future__ import annotations
7import re
8from collections import defaultdict
9from itertools import chain
10from typing import TYPE_CHECKING, Literal, overload
12from django.core.cache import cache
14from weblate.glossary.models import (
15 fetch_glossary_terms,
16 get_glossary_terms,
17 render_glossary_units_tsv,
18)
19from weblate.utils.errors import add_breadcrumb
21from .base import (
22 BatchMachineTranslation,
23 MachineryRateLimitError,
24 MachineTranslationError,
25)
26from .forms import AzureOpenAIMachineryForm, OpenAIMachineryForm
28if TYPE_CHECKING: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true
29 from collections.abc import Iterable
31 from openai import OpenAI
33 from weblate.trans.models import Unit
35 from .base import (
36 DownloadMultipleTranslations,
37 )
40PROMPT = """
41You are a highly skilled translation assistant, adept at translating text
42from language '{source_language}'
43to language '{target_language}'
44with precision and nuance.
45{persona}
46{style}
47You always reply with translated string only.
48You do not include transliteration.
49{separator}
50{placeables}
51{glossary}
52"""
53SEPARATOR = "\n==WEBLATE_PART==\n"
54SEPARATOR_RE = re.compile(r"\n *==WEBLATE_PART== *\n")
55SEPARATOR_PROMPT = f"""
56You receive an input as strings separated by {SEPARATOR} and
57your answer separates strings by {SEPARATOR}.
58"""
59REPHRASE_PROMPT = f"""
60You receive an input as the source and existing translation strings separated
61by {SEPARATOR} and you answer three rephrased translation strings separated by
62{SEPARATOR}.
63"""
64GLOSSARY_PROMPT = """
65Use the following glossary during the translation:
66{}
67"""
68PLACEABLES_PROMPT = """
69You treat strings like {placeable_1} or {placeable_2} as placeables for user input and keep them intact.
70"""
73class BaseOpenAITranslation(BatchMachineTranslation):
74 max_score = 90
75 request_timeout = 60
76 client: OpenAI
77 glossary_support = True
79 def __init__(self, settings=None) -> None:
80 super().__init__(settings)
82 def is_supported(self, source_language, target_language) -> bool:
83 return True
85 def format_prompt_part(self, name: Literal["style", "persona"]):
86 text = self.settings[name]
87 text = text.strip()
88 if text and not text.endswith("."):
89 text = f"{text}."
90 return text
92 def _get_prompt(
93 self,
94 source_language: str,
95 target_language: str,
96 texts: list[str],
97 units: list[Unit | None],
98 *,
99 rephrase: bool = False,
100 ) -> str:
101 glossary = ""
103 if any(units):
104 fetch_glossary_terms([unit for unit in units if unit is not None])
105 glossary = render_glossary_units_tsv(
106 chain.from_iterable(
107 get_glossary_terms(unit, include_variants=False)
108 for unit in units
109 if unit is not None
110 )
111 )
112 if glossary:
113 glossary = GLOSSARY_PROMPT.format(glossary)
115 separator = ""
116 if rephrase:
117 separator = REPHRASE_PROMPT
118 elif len(units) > 1:
119 separator = SEPARATOR_PROMPT
121 placeables = ""
122 if any(self.replacement_start in text for text in texts):
123 placeables = PLACEABLES_PROMPT.format(
124 placeable_1=self.format_replacement(0, -1, "", None),
125 placeable_2=self.format_replacement(123, -1, "", None),
126 )
128 return PROMPT.format(
129 source_language=source_language,
130 target_language=target_language,
131 persona=self.format_prompt_part("persona"),
132 style=self.format_prompt_part("style"),
133 glossary=glossary,
134 separator=separator,
135 placeables=placeables,
136 )
138 def download_multiple_translations(
139 self,
140 source_language,
141 target_language,
142 sources: list[tuple[str, Unit | None]],
143 user=None,
144 threshold: int = 75,
145 ) -> DownloadMultipleTranslations:
146 rephrase: list[tuple[str, Unit]] = []
147 texts: list[str] = []
148 units: list[Unit | None] = []
150 # Separate rephrasing and new translations
151 for text, unit in sources:
152 if (
153 unit is not None
154 and unit.translated
155 and not unit.readonly
156 and all(unit.get_target_plurals())
157 ):
158 rephrase.append((text, unit))
159 else:
160 texts.append(text)
161 units.append(unit)
163 # Collect results
164 result: DownloadMultipleTranslations = defaultdict(list)
166 # Fetch rephrasing each string separately
167 if rephrase:
168 for text, unit in rephrase:
169 self._download(
170 result,
171 source_language,
172 target_language,
173 [text],
174 [unit],
175 rephrase=True,
176 )
178 # Fetch translations in batch
179 if texts:
180 self._download(result, source_language, target_language, texts, units)
182 return result
184 @overload
185 def _download( 185 ↛ anywhereline 185 didn't jump anywhere: it always raised an exception.
186 self,
187 result: DownloadMultipleTranslations,
188 source_language,
189 target_language,
190 texts: list[str],
191 units: list[Unit],
192 *,
193 rephrase: Literal[True],
194 ): ...
195 @overload
196 def _download( 196 ↛ anywhereline 196 didn't jump anywhere: it always raised an exception.
197 self,
198 result: DownloadMultipleTranslations,
199 source_language,
200 target_language,
201 texts: list[str],
202 units: list[Unit | None],
203 ): ...
204 def _download(
205 self,
206 result: DownloadMultipleTranslations,
207 source_language,
208 target_language,
209 texts,
210 units,
211 *,
212 rephrase=False,
213 ):
214 from openai import RateLimitError
215 from openai.types.chat import (
216 ChatCompletionSystemMessageParam,
217 ChatCompletionUserMessageParam,
218 )
220 prompt = self._get_prompt(
221 source_language, target_language, texts, units, rephrase=rephrase
222 )
223 content = SEPARATOR.join(texts if not rephrase else [*texts, units[0].target])
224 add_breadcrumb("openai", "prompt", prompt=prompt)
225 add_breadcrumb("openai", "chat", content=content)
227 messages: Iterable[
228 ChatCompletionSystemMessageParam | ChatCompletionUserMessageParam
229 ] = [
230 ChatCompletionSystemMessageParam(role="system", content=prompt),
231 ChatCompletionUserMessageParam(
232 role="user",
233 content=content,
234 ),
235 ]
237 try:
238 response = self.client.chat.completions.create(
239 model=self.get_model(),
240 messages=messages,
241 )
242 except RateLimitError as error:
243 if not isinstance(error.body, dict) or not (
244 message := error.body.get("message")
245 ):
246 message = error.message
247 raise MachineryRateLimitError(message) from error
249 translations_string = response.choices[0].message.content
250 add_breadcrumb("openai", "response", translations_string=translations_string)
251 if translations_string is None:
252 self.report_error(
253 "Blank assistant reply",
254 extra_log=translations_string,
255 message=True,
256 )
257 msg = "Blank assistant reply"
258 raise MachineTranslationError(msg)
260 # Ignore extra whitespace in response as OpenAI can be creative in that
261 # (see https://github.com/WeblateOrg/weblate/issues/12456)
262 translations = SEPARATOR_RE.split(translations_string)
263 if not rephrase and len(translations) != len(texts):
264 self.report_error(
265 "Failed to parse assistant reply",
266 extra_log=translations_string,
267 message=True,
268 )
269 msg = f"Could not parse assistant reply, expected={len(texts)}, received={len(translations)}"
270 raise MachineTranslationError(msg)
272 for index, translation in enumerate(translations):
273 text = texts[index if not rephrase else 0]
274 result[text].append(
275 {
276 "text": translation,
277 "quality": self.max_score,
278 "service": self.name,
279 "source": text,
280 }
281 )
283 def get_model(self) -> str:
284 raise NotImplementedError
287class OpenAITranslation(BaseOpenAITranslation):
288 name = "OpenAI"
290 settings_form = OpenAIMachineryForm
292 def __init__(self, settings=None) -> None:
293 from openai import OpenAI
295 super().__init__(settings)
296 self.client = OpenAI(
297 api_key=self.settings["key"],
298 timeout=self.request_timeout,
299 base_url=self.settings.get("base_url") or None,
300 )
301 self._models: set[str] | None = None
303 def get_model(self) -> str:
304 if self._models is None:
305 cache_key = self.get_cache_key("models")
306 models_cache = cache.get(cache_key)
307 if models_cache is not None:
308 # hiredis-py 3 makes list from set
309 self._models = set(models_cache)
310 else:
311 self._models = {model.id for model in self.client.models.list()}
312 cache.set(cache_key, self._models, 3600)
314 if self.settings["model"] in self._models:
315 return self.settings["model"]
316 if self.settings["model"] == "auto":
317 for model, _name in self.settings_form.MODEL_CHOICES:
318 if model == "auto":
319 continue
320 if model in self._models:
321 return model
322 if self.settings["model"] == "custom":
323 return self.settings["custom_model"]
325 msg = f"Unsupported model: {self.settings['model']}"
326 raise MachineTranslationError(msg)
329class AzureOpenAITranslation(BaseOpenAITranslation):
330 name = "Azure OpenAI"
331 settings_form = AzureOpenAIMachineryForm
333 def __init__(self, settings=None) -> None:
334 from openai import AzureOpenAI
336 super().__init__(settings)
337 self.client = AzureOpenAI(
338 api_key=self.settings["key"],
339 api_version="2024-06-01",
340 timeout=self.request_timeout,
341 azure_endpoint=self.settings.get("azure_endpoint") or "",
342 azure_deployment=self.settings["deployment"],
343 )
345 def get_model(self) -> str:
346 return self.settings["deployment"]