From e377c965e9869364c223bcd3f92390e756764e3e Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 4 Jun 2026 20:09:45 +0300 Subject: [PATCH] =?UTF-8?q?032:=20Phase=204=20US2=20=E2=80=94=20Translate?= =?UTF-8?q?=20plugin=20fully=20async?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T022-T028: All translate methods async. - _llm_async_http.py created (httpx.AsyncClient+asyncio.sleep) - Old _llm_http.py and preview_llm_client.py preserved (tombstone later) - superset_executor, preview, executor, run_source, llm_call all async RATIONALE: httpx.AsyncClient + asyncio.sleep instead of time.sleep. REJECTED: AsyncOpenAI SDK — doesn't support custom base_url. --- backend/src/plugins/translate/_batch_proc.py | 8 +- .../src/plugins/translate/_llm_async_http.py | 243 ++++++++++++++++++ backend/src/plugins/translate/_llm_call.py | 44 ++-- backend/src/plugins/translate/_run_service.py | 4 +- backend/src/plugins/translate/_run_source.py | 6 +- backend/src/plugins/translate/executor.py | 38 +-- backend/src/plugins/translate/preview.py | 6 +- .../src/plugins/translate/preview_executor.py | 14 +- .../plugins/translate/superset_executor.py | 26 +- 9 files changed, 316 insertions(+), 73 deletions(-) create mode 100644 backend/src/plugins/translate/_llm_async_http.py diff --git a/backend/src/plugins/translate/_batch_proc.py b/backend/src/plugins/translate/_batch_proc.py index 13c69c79..c92dd2f0 100644 --- a/backend/src/plugins/translate/_batch_proc.py +++ b/backend/src/plugins/translate/_batch_proc.py @@ -46,7 +46,7 @@ class BatchProcessingService: # @PRE job and batch_rows are valid. # @POST TranslationBatch and TranslationRecord rows are created. # @SIDE_EFFECT LLM API call; DB writes. - def process_batch( + async def process_batch( self, job: TranslationJob, run_id: str, batch_index: int, batch_rows: list[dict[str, Any]], dict_snapshot_hash: str | None = None, config_hash: str | None = None, @@ -76,7 +76,7 @@ class BatchProcessingService: # Count cache hits: pre_rows that have _cached_lang_values (served from cache, not same-lang/approved) result["cache_hits"] = sum(1 for r in pre_rows if r.get("_cached_lang_values")) if llm_rows: - llm_res = self._process_llm(job, run_id, llm_rows, dict_matches, bid, tls) + llm_res = await self._process_llm(job, run_id, llm_rows, dict_matches, bid, tls) for k in ("successful", "failed", "skipped", "retries"): result[k] += llm_res.get(k, 0) @@ -205,7 +205,7 @@ class BatchProcessingService: count += 1 return count - def _process_llm(self, job, run_id, rows_for_llm, dict_matches, bid, tls): + async def _process_llm(self, job, run_id, rows_for_llm, dict_matches, bid, tls): # Resolve provider token config (DB values take priority over PROVIDER_DEFAULTS) token_config = {"model": None, "context_window": None, "max_output_tokens": None} if job.provider_id: @@ -239,7 +239,7 @@ class BatchProcessingService: }, ) - result = self._llm_service.call_llm_for_batch( + result = await self._llm_service.call_llm_for_batch( job=job, run_id=run_id, batch_rows=rows_for_llm, dict_matches=dict_matches, batch_id=bid, max_tokens=tb["max_output_needed"], diff --git a/backend/src/plugins/translate/_llm_async_http.py b/backend/src/plugins/translate/_llm_async_http.py new file mode 100644 index 00000000..3861009a --- /dev/null +++ b/backend/src/plugins/translate/_llm_async_http.py @@ -0,0 +1,243 @@ +# #region LLMAsyncHttpClient [C:4] [TYPE Module] [SEMANTICS translate, llm, http, openai, async, retry, rate-limit] +# @BRIEF Async HTTP client for OpenAI-compatible LLM API calls using httpx.AsyncClient +# with rate-limit handling and structured output fallback. +# @LAYER Infrastructure +# @RELATION DEPENDS_ON -> [EXT:httpx:AsyncClient] +# @PRE Valid API endpoint, key, model, and prompt. +# @POST Returns (response text, finish_reason) tuple. +# @SIDE_EFFECT Async HTTP POST to LLM API with optional retry on 429. +# @RATIONALE Async migration of _llm_http.py to use httpx.AsyncClient instead of sync requests. +# Uses asyncio.sleep for 429 backoff instead of time.sleep. +# @REJECTED Keeping sync requests.post — would block async event loop during LLM calls. + +import asyncio +import os +from typing import Any + +import httpx + +from ...core.logger import logger + +# Module-level httpx client, lazily initialized for connection reuse +_http_client: httpx.AsyncClient | None = None + +# Default provider and max_tokens constants +DEFAULT_PROVIDER_TYPE: str = "openai" +DEFAULT_MAX_TOKENS: int = 8192 + + +# #region _get_verify [C:1] [TYPE Function] [SEMANTICS translate, ssl, verify] +# @BRIEF Resolve SSL verification path from LLM_SSL_VERIFY env var. +# @RATIONALE Используем capath=/etc/ssl/certs/ вместо cafile, потому что +# OpenSSL 3.x не использует intermediate CA сертификаты из cafile для +# построения цепочки (verify code 20). capath с хеш-симлинками работает +# корректно (verify code 0). +# @REJECTED cafile отвергнут — OpenSSL 3.x не использует intermediate CA +# из единого bundle-файла. Только capath с хеш-симлинками даёт code 0. +# @POST Returns path to /etc/ssl/certs/ when enabled, False when disabled. +def _get_verify() -> str | bool: + raw = os.getenv("LLM_SSL_VERIFY", "true").strip().lower() + if raw in ("false", "0", "no", "off"): + return False + return "/etc/ssl/certs/" +# #endregion _get_verify + + +# #region _get_http_client [C:1] [TYPE Function] +# @BRIEF Get or create the module-level httpx.AsyncClient singleton. +# @POST Returns httpx.AsyncClient with SSL verify and 180s timeout. +async def _get_http_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None: + _http_client = httpx.AsyncClient( + verify=_get_verify(), + timeout=httpx.Timeout(180.0), + ) + return _http_client +# #endregion _get_http_client + + +# #region call_openai_compatible [C:3] [TYPE Function] [SEMANTICS translate, llm, http, openai, async] +# @BRIEF Call OpenAI-compatible API asynchronously with rate-limit handling and structured output fallback. +# @PRE Valid API endpoint, key, model, and prompt. +# @POST Returns (response text, finish_reason). +# @SIDE_EFFECT Async HTTP POST to LLM API. +async def call_openai_compatible( + base_url: str, + api_key: str, + model: str, + prompt: str, + provider_type: str = "openai", + max_tokens: int = 8192, + disable_reasoning: bool = False, +) -> tuple[str, str | None]: + """Call OpenAI-compatible API for batch translation (async).""" + if not base_url: + raise ValueError("LLM provider has no base_url configured") + + url = f"{base_url.rstrip('/')}/chat/completions" + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + system_content = ( + "You are a database content translation assistant. " + "Translate the provided text accurately, preserving data semantics. " + "Respond directly with ONLY the JSON result. " + "Do NOT include any reasoning, thinking, chain-of-thought, analysis, " + "or explanation. Output ONLY valid JSON." + ) + + payload: dict[str, Any] = { + "model": model, + "messages": [ + {"role": "system", "content": system_content}, + {"role": "user", "content": prompt}, + ], + "temperature": 0.1, + "max_tokens": max_tokens, + } + + if provider_type in ("openai", "openai_compatible", "kilo", "openrouter", "litellm"): + if not disable_reasoning: + payload["response_format"] = {"type": "json_object"} + + if disable_reasoning: + if provider_type not in ("kilo", "openrouter", "litellm"): + payload["reasoning_effort"] = "none" + payload["max_tokens"] = max_tokens + + logger.reason( + f"LLM request url={base_url} model={payload.get('model')} " + f"provider_type={provider_type} " + f"response_format={'yes' if 'response_format' in payload else 'no'} " + f"prompt_len={len(prompt)}" + ) + + response, response_text = await _do_http_request(url, headers, payload) + await _handle_response_format_fallback(response, response_text, payload, url, headers) + + if not response.ok: + logger.explore( + f"LLM API error status={response.status_code} " + f"model={payload.get('model')} " + f"body={response_text[:2000]}" + ) + response.raise_for_status() + data = response.json() + + choices = data.get("choices", []) + if not choices: + logger.explore("LLM returned no choices", extra={ + "src": "executor", "response_keys": list(data.keys()), + "response_preview": str(data)[:2000], + }) + raise ValueError("LLM returned no choices") + + try: + finish_reason = choices[0].get("finish_reason") or "none" + msg = choices[0].get("message") or {} + except (TypeError, AttributeError) as e: + logger.explore("TypeError processing LLM response choices", extra={ + "src": "executor_diag", "error": str(e), + "choices_0_type": type(choices[0]).__name__ if choices else "N/A", + "choices_0_repr": repr(choices[0])[:2000] if choices else "N/A", + "data_type": type(data).__name__, "data_preview": str(data)[:2000], + }) + raise ValueError(f"LLM response processing failed: {e}") + + # Log provider token usage for batch sizing calibration + usage = data.get("usage") or {} + if usage: + logger.reason( + "LLM provider usage", + { + "prompt_tokens": usage.get("prompt_tokens"), + "completion_tokens": usage.get("completion_tokens"), + "total_tokens": usage.get("total_tokens"), + "finish_reason": finish_reason, + "max_tokens_sent": max_tokens, + "chars_sent": len(prompt), + }, + ) + + refusal = msg.get("refusal") if isinstance(msg, dict) else None + if refusal: + logger.explore("LLM refused to respond", extra={ + "src": "executor", "refusal": str(refusal)[:500], "finish_reason": finish_reason, + }) + raise ValueError(f"LLM refused to respond: {refusal}") + + content = msg.get("content") if isinstance(msg, dict) else "" + if not content and isinstance(msg, dict): + content = msg.get("content") or "" + + logger.reason( + f"LLM response finish_reason={finish_reason} content_len={len(content)} " + f"msg_keys={list(msg.keys()) if isinstance(msg, dict) else []}" + ) + if not content: + logger.explore("LLM returned empty content", extra={ + "src": "executor", "finish_reason": finish_reason, + "msg_keys": list(msg.keys()) if isinstance(msg, dict) else [], + "response_preview": str(data)[:2000], + }) + raise ValueError("LLM returned empty content") + + return content, finish_reason +# #endregion call_openai_compatible + + +# #region _do_http_request [C:1] [TYPE Function] +async def _do_http_request(url: str, headers: dict, payload: dict) -> tuple[httpx.Response, str]: + """Make async HTTP POST with rate-limit (429) retry handling.""" + client = await _get_http_client() + _max_retry_429 = 3 + _retry_count_429 = 0 + while _retry_count_429 < _max_retry_429: + response = await client.post(url, headers=headers, json=payload) + response_text = response.text + if response.status_code == 429: + _retry_count_429 += 1 + retry_after = response.headers.get("Retry-After") + if retry_after: + try: + wait = int(retry_after) + except (ValueError, TypeError): + wait = 2 ** _retry_count_429 + else: + wait = 2 ** _retry_count_429 + logger.explore(f"Rate limited (429), retry {_retry_count_429}/{_max_retry_429} after {wait}s", + extra={"src": "executor", "retry_after": retry_after, "wait": wait}) + await asyncio.sleep(wait) + if _retry_count_429 >= _max_retry_429: + break + else: + break + return response, response_text +# #endregion _do_http_request + + +# #region _handle_response_format_fallback [C:1] [TYPE Function] +async def _handle_response_format_fallback( + response: httpx.Response, response_text: str, payload: dict, url: str, headers: dict, +) -> None: + """Handle 400 errors from structured_outputs not being supported.""" + _patterns = ("response_format", "structured_outputs", "structured", "json_object") + if ( + not response.ok + and response.status_code == 400 + and any(p in (response_text or "").lower() for p in _patterns) + ): + client = await _get_http_client() + logger.explore("Structured outputs not supported, retrying without response_format", + extra={"src": "executor"}) + payload.pop("response_format", None) + new_response = await client.post(url, headers=headers, json=payload) + # Mutate the original response object with new data + response.status_code = new_response.status_code + response._content = new_response.content + response.encoding = new_response.encoding + response.headers = new_response.headers +# #endregion _handle_response_format_fallback +# #endregion LLMAsyncHttpClient diff --git a/backend/src/plugins/translate/_llm_call.py b/backend/src/plugins/translate/_llm_call.py index e10ebb90..fa83cd88 100644 --- a/backend/src/plugins/translate/_llm_call.py +++ b/backend/src/plugins/translate/_llm_call.py @@ -16,9 +16,9 @@ # _llm_parse.py to meet INV_7 module limit (< 400 lines). # @REJECTED Single monolithic call_llm_for_batch at 327 lines — split into focused sub-methods. +import asyncio from datetime import UTC, datetime import json -import time from typing import Any import uuid @@ -28,7 +28,7 @@ from ...core.logger import belief_scope, logger from ...models.translate import TranslationBatch, TranslationJob, TranslationLanguage, TranslationRecord from ...services.llm_prompt_templates import render_prompt from ...services.llm_provider import LLMProviderService -from ._llm_http import call_openai_compatible +from ._llm_async_http import call_openai_compatible from ._llm_parse import parse_llm_response from ._utils import _enforce_dictionary from .preview import DEFAULT_EXECUTION_PROMPT_TEMPLATE @@ -49,7 +49,7 @@ class LLMTranslationService: # @PRE job has valid provider_id. batch_rows is non-empty. # @POST Returns dict with successful/failed/skipped counts. # @SIDE_EFFECT HTTP call to LLM provider; DB writes. - def call_llm_for_batch( + async def call_llm_for_batch( self, job: TranslationJob, run_id: str, batch_rows: list[dict[str, Any]], dict_matches: list[dict[str, Any]], batch_id: str, max_tokens: int = 8192, _recursion_depth: int = 0, @@ -68,7 +68,7 @@ class LLMTranslationService: target_languages = self._resolve_target_languages(job) prompt = self._build_prompt(job, batch_rows, dictionary_section, target_languages) - llm_response, finish_reason, retries, last_error = self._call_llm_with_retry( + llm_response, finish_reason, retries, last_error = await self._call_llm_with_retry( job, prompt, batch_id, max_tokens, ) if llm_response is None: @@ -96,7 +96,7 @@ class LLMTranslationService: f"Retrying only {len(remaining)}/{len(batch_rows)} missing rows", {"batch_id": batch_id, "recovered": len(recovered_ids), "remaining": len(remaining)}, ) - return self._retry_missing_rows( + return await self._retry_missing_rows( job, run_id, remaining, dict_matches, batch_id, max_tokens, _recursion_depth, ) @@ -116,8 +116,8 @@ class LLMTranslationService: "depth": _recursion_depth, }, ) - return self._split_and_retry(job, run_id, batch_rows, dict_matches, - batch_id, max_tokens, _recursion_depth, retries) + return await self._split_and_retry(job, run_id, batch_rows, dict_matches, + batch_id, max_tokens, _recursion_depth, retries) logger.explore("Truncation recursion depth exceeded", {"batch_id": batch_id, "depth": _recursion_depth}) try: @@ -187,7 +187,7 @@ class LLMTranslationService: # #region _call_llm_with_retry [C:3] [TYPE Function] [SEMANTICS translate, llm, retry] # @BRIEF Call LLM with retry loop (max 3 attempts, exponential backoff). # @SIDE_EFFECT HTTP calls to LLM provider on each attempt. - def _call_llm_with_retry(self, job, prompt, batch_id, max_tokens): + async def _call_llm_with_retry(self, job, prompt, batch_id, max_tokens): llm_response = None last_error = None retries = 0 @@ -195,7 +195,7 @@ class LLMTranslationService: logger.reason(f"LLM retry loop start", {"batch_id": batch_id, "max_retries": MAX_RETRIES_PER_BATCH, "prompt_len": len(prompt)}) for attempt in range(1, MAX_RETRIES_PER_BATCH + 1): try: - llm_response, finish_reason = self.call_llm(job, prompt, max_tokens=max_tokens) + llm_response, finish_reason = await self.call_llm(job, prompt, max_tokens=max_tokens) logger.reason( f"LLM call succeeded (attempt {attempt})", { "batch_id": batch_id, "finish_reason": finish_reason, @@ -208,7 +208,7 @@ class LLMTranslationService: retries += 1 logger.explore(f"LLM call failed (attempt {attempt})", {"batch_id": batch_id, "error": last_error}) if attempt < MAX_RETRIES_PER_BATCH: - time.sleep(2 ** attempt) + await asyncio.sleep(2 ** attempt) if llm_response is None: logger.explore(f"All LLM retries exhausted", {"batch_id": batch_id, "retries": retries, "last_error": last_error}) return llm_response, finish_reason, retries, last_error @@ -233,7 +233,7 @@ class LLMTranslationService: # #region _split_and_retry [C:3] [TYPE Function] [SEMANTICS translate, llm, split, retry] # @BRIEF Binary-split a batch and retry each half recursively. # @SIDE_EFFECT Creates two child TranslationBatch rows; recursive LLM calls. - def _split_and_retry(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens, depth, retries): + async def _split_and_retry(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens, depth, retries): mid = len(batch_rows) // 2 logger.explore("LLM output truncated — splitting batch", {"batch_id": batch_id, "batch_size": len(batch_rows), "split_at": mid, "depth": depth}) @@ -257,10 +257,10 @@ class LLMTranslationService: self.db.add_all([left_batch, right_batch]) self.db.flush() - left = self.call_llm_for_batch(job, run_id, batch_rows[:mid], dict_matches, - left_batch.id, max_tokens, depth + 1) - right = self.call_llm_for_batch(job, run_id, batch_rows[mid:], dict_matches, - right_batch.id, max_tokens, depth + 1) + left = await self.call_llm_for_batch(job, run_id, batch_rows[:mid], dict_matches, + left_batch.id, max_tokens, depth + 1) + right = await self.call_llm_for_batch(job, run_id, batch_rows[mid:], dict_matches, + right_batch.id, max_tokens, depth + 1) # Finalise child batch stats left_batch.completed_at = datetime.now(UTC) @@ -355,7 +355,7 @@ class LLMTranslationService: # @PRE missing_rows is a subset of the original batch rows, or empty. # @POST Returns dict with successful/failed/skipped counts from the sub-batch. # @SIDE_EFFECT Creates TranslationBatch for the retry sub-batch; may recurse. - def _retry_missing_rows(self, job, run_id, missing_rows, dict_matches, _batch_id, max_tokens, depth): + async def _retry_missing_rows(self, job, run_id, missing_rows, dict_matches, _batch_id, max_tokens, depth): if not missing_rows: return {"successful": 0, "failed": 0, "skipped": 0, "retries": 0} @@ -367,7 +367,7 @@ class LLMTranslationService: self.db.add(sub_batch) self.db.flush() - result = self.call_llm_for_batch( + result = await self.call_llm_for_batch( job, run_id, missing_rows, dict_matches, sub_batch.id, max_tokens, depth, ) @@ -483,7 +483,7 @@ class LLMTranslationService: # #region call_llm [C:3] [TYPE Function] [SEMANTICS translate, llm, call] # @BRIEF Route to provider-specific LLM call implementation. - def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> tuple[str, str | None]: + async def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> tuple[str, str | None]: with belief_scope("LLMTranslationService.call_llm"): if not job.provider_id: raise ValueError("Job has no LLM provider configured") @@ -508,7 +508,7 @@ class LLMTranslationService: if provider_type not in ("openai", "openai_compatible", "openrouter", "kilo", "litellm"): raise ValueError(f"Unsupported provider type '{provider_type}'") - result = call_openai_compatible( + result = await call_openai_compatible( base_url=provider.base_url, api_key=api_key, model=model, prompt=prompt, provider_type=provider_type, max_tokens=max_tokens, disable_reasoning=disable_reasoning, ) @@ -522,10 +522,10 @@ class LLMTranslationService: # #endregion call_llm # #region call_openai_compatible [C:1] [TYPE Function] [SEMANTICS translate, llm, compat] - # @BRIEF Backward-compat delegation to _llm_http.call_openai_compatible. + # @BRIEF Backward-compat delegation to _llm_async_http.call_openai_compatible. @staticmethod - def call_openai_compatible(*a, **kw): - return call_openai_compatible(*a, **kw) + async def call_openai_compatible(*a, **kw): + return await call_openai_compatible(*a, **kw) # #endregion call_openai_compatible # #region _parse_llm_response [C:1] [TYPE Function] [SEMANTICS translate, llm, compat] diff --git a/backend/src/plugins/translate/_run_service.py b/backend/src/plugins/translate/_run_service.py index 0cbd37c5..09306c67 100644 --- a/backend/src/plugins/translate/_run_service.py +++ b/backend/src/plugins/translate/_run_service.py @@ -39,8 +39,8 @@ class RunExecutionService: self._preview_edits_cache: dict[str, dict[str, str]] | None = None # -- Source fetching (thin wrappers) -- - def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]: - return fetch_source_rows(self.db, self.config_manager, job_id, run_id) + async def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]: + return await fetch_source_rows(self.db, self.config_manager, job_id, run_id) @staticmethod def _extract_chart_data_rows(response): diff --git a/backend/src/plugins/translate/_run_source.py b/backend/src/plugins/translate/_run_source.py index f5c86859..ebcb0a53 100644 --- a/backend/src/plugins/translate/_run_source.py +++ b/backend/src/plugins/translate/_run_source.py @@ -25,7 +25,7 @@ MAX_ROWS_PER_RUN = 10000 # #region fetch_source_rows [C:3] [TYPE Function] [SEMANTICS translate, source, fetch] # @BRIEF Fetch source rows from Superset datasource or preview session fallback. -def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, run_id: str) -> list[dict[str, Any]]: +async def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, run_id: str) -> list[dict[str, Any]]: """Fetch source rows from Superset datasource or preview session fallback.""" with belief_scope("RunSourceFetcher.fetch_source_rows"): job = db.query(TranslationJob).filter(TranslationJob.id == job_id).first() @@ -45,7 +45,7 @@ def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, r if env_config: from ...core.superset_client import SupersetClient client = SupersetClient(env_config) - dataset_detail = client.get_dataset_detail(int(job.source_datasource_id)) + dataset_detail = await client.get_dataset_detail(int(job.source_datasource_id)) query_context = client.build_dataset_preview_query_context( dataset_id=int(job.source_datasource_id), dataset_record=dataset_detail, template_params={}, effective_filters=[], @@ -59,7 +59,7 @@ def fetch_source_rows(db: Session, config_manager: ConfigManager, job_id: str, r form_data = query_context.get("form_data", {}) form_data.pop("query_mode", None) - response = client.network.request( + response = await client.network.request( method="POST", endpoint="/api/v1/chart/data", data=json.dumps(query_context), headers={"Content-Type": "application/json"}, ) diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py index 1765c88a..fde4fa48 100644 --- a/backend/src/plugins/translate/executor.py +++ b/backend/src/plugins/translate/executor.py @@ -59,7 +59,7 @@ class TranslationExecutor: # @PRE run is in PENDING or RUNNING status with valid job config. # @POST Run is populated with batches and records. # @SIDE_EFFECT LLM API calls; DB batch writes. - def execute_run( + async def execute_run( self, run: TranslationRun, llm_progress_callback: Callable[[str, int, int, int], None] | None = None, language_stats_map: dict[str, TranslationRunLanguageStats] | None = None, @@ -70,7 +70,7 @@ class TranslationExecutor: raise ValueError(f"Job '{run.job_id}' not found for run '{run.id}'") logger.reason("Starting translation execution", {"run_id": run.id, "job_id": job.id}) - run, source_rows, job = self._prepare_run(run, job) + run, source_rows, job = await self._prepare_run(run, job) if run.status != "RUNNING": return run @@ -78,11 +78,11 @@ class TranslationExecutor: run.total_records = len(source_rows) logger.reason(f"Processing {len(batches)} batches", {"run_id": run.id, "total_rows": run.total_records}) - return self._process_batches(run, job, batches, target_languages, language_stats_map) + return await self._process_batches(run, job, batches, target_languages, language_stats_map) # endregion execute_run # region _prepare_run [C:2] [TYPE Function] - def _prepare_run(self, run: TranslationRun, job: TranslationJob) -> tuple: + async def _prepare_run(self, run: TranslationRun, job: TranslationJob) -> tuple: """Prepare run: load preview edits, fetch source rows, filter new keys.""" self._load_preview_edits(job.id) run.status = "RUNNING" @@ -93,7 +93,7 @@ class TranslationExecutor: if run.config_snapshot and isinstance(run.config_snapshot, dict): full_translation = run.config_snapshot.get("full_translation", False) - source_rows = self._fetch_source_rows(job.id, run.id) + source_rows = await self._fetch_source_rows(job.id, run.id) if not source_rows: logger.explore("No source rows to translate", {"run_id": run.id}) run.status = "COMPLETED" @@ -125,13 +125,13 @@ class TranslationExecutor: # endregion _prepare_batches # region _process_batches [C:3] [TYPE Function] - def _process_batches(self, run: TranslationRun, job: TranslationJob, batches: list, - target_languages: list[str], - language_stats_map: dict[str, TranslationRunLanguageStats] | None = None) -> TranslationRun: + async def _process_batches(self, run: TranslationRun, job: TranslationJob, batches: list, + target_languages: list[str], + language_stats_map: dict[str, TranslationRunLanguageStats] | None = None) -> TranslationRun: """Process all batches: execute, insert to target, check cancellation.""" successful_records = failed_records = skipped_records = cache_hits = 0 for batch_idx, batch_rows in enumerate(batches): - batch_result = self._process_batch( + batch_result = await self._process_batch( job=job, run_id=run.id, batch_index=batch_idx, batch_rows=batch_rows, dict_snapshot_hash=run.dict_snapshot_hash, config_hash=run.config_hash, ) @@ -187,8 +187,8 @@ class TranslationExecutor: return run # -- Delegation methods (thin wrappers for test-patch compatibility) -- - def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]: - return self._run_service._fetch_source_rows(job_id, run_id) + async def _fetch_source_rows(self, job_id: str, run_id: str) -> list[dict[str, Any]]: + return await self._run_service._fetch_source_rows(job_id, run_id) def _filter_new_keys(self, job, run_id: str, source_rows: list) -> list: return self._run_service._filter_new_keys(job, run_id, source_rows) @@ -241,9 +241,9 @@ class TranslationExecutor: return AdaptiveBatchSizer(self.db, self.config_manager).auto_size_batches( job, source_rows, target_languages, provider_info) - def _process_batch(self, job, run_id, batch_index, batch_rows, dict_snapshot_hash=None, config_hash=None) -> dict: + async def _process_batch(self, job, run_id, batch_index, batch_rows, dict_snapshot_hash=None, config_hash=None) -> dict: from ._batch_proc import BatchProcessingService - return BatchProcessingService(self.db, self.config_manager).process_batch( + return await BatchProcessingService(self.db, self.config_manager).process_batch( job, run_id, batch_index, batch_rows, dict_snapshot_hash, config_hash, preview_edits_cache=self._preview_edits_cache) @@ -251,19 +251,19 @@ class TranslationExecutor: from ._batch_proc import BatchProcessingService BatchProcessingService(self.db, self.config_manager).insert_batch_to_target(job, batch_id, run_id) - def _call_llm_for_batch(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens=8192, _recursion_depth=0) -> dict: + async def _call_llm_for_batch(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens=8192, _recursion_depth=0) -> dict: from ._llm_call import LLMTranslationService - return LLMTranslationService(self.db).call_llm_for_batch( + return await LLMTranslationService(self.db).call_llm_for_batch( job, run_id, batch_rows, dict_matches, batch_id, max_tokens, _recursion_depth) - def _call_llm(self, job, prompt, max_tokens=8192) -> tuple: + async def _call_llm(self, job, prompt, max_tokens=8192) -> tuple: from ._llm_call import LLMTranslationService - return LLMTranslationService(self.db).call_llm(job, prompt, max_tokens) + return await LLMTranslationService(self.db).call_llm(job, prompt, max_tokens) @staticmethod - def _call_openai_compatible(base_url, api_key, model, prompt, provider_type="openai", max_tokens=8192, disable_reasoning=False) -> tuple: + async def _call_openai_compatible(base_url, api_key, model, prompt, provider_type="openai", max_tokens=8192, disable_reasoning=False) -> tuple: from ._llm_call import LLMTranslationService - return LLMTranslationService.call_openai_compatible(base_url, api_key, model, prompt, provider_type, max_tokens, disable_reasoning) + return await LLMTranslationService.call_openai_compatible(base_url, api_key, model, prompt, provider_type, max_tokens, disable_reasoning) @staticmethod def _parse_llm_response(response_text, expected_count, target_languages=None, finish_reason=None) -> dict: diff --git a/backend/src/plugins/translate/preview.py b/backend/src/plugins/translate/preview.py index 3dae82c4..475fa39c 100644 --- a/backend/src/plugins/translate/preview.py +++ b/backend/src/plugins/translate/preview.py @@ -65,7 +65,7 @@ class TranslationPreview: # region preview_rows [TYPE Function] # @PURPOSE: Fetch sample rows, send to LLM, create preview session with per-language records. # @SIDE_EFFECT Fetches data from Superset; calls LLM; creates DB rows. - def preview_rows( + async def preview_rows( self, job_id: str, sample_size: int = 10, @@ -91,7 +91,7 @@ class TranslationPreview: config_hash = self._executor.compute_config_hash(job) dict_snapshot_hash = self._executor.compute_dict_snapshot_hash(job_id) - source_rows = self._executor.fetch_sample_rows(job=job, sample_size=sample_size, env_id=env_id) + source_rows = await self._executor.fetch_sample_rows(job=job, sample_size=sample_size, env_id=env_id) if not source_rows: raise ValueError("No rows returned from datasource for preview") @@ -116,7 +116,7 @@ class TranslationPreview: meta["_detected_lang"] = detect_language(meta["source_text"], detector) t_llm = time.monotonic() - llm_response = self._executor.call_llm( + llm_response = await self._executor.call_llm( job=job, prompt=prompt_data["prompt"], max_tokens=token_budget["max_output_needed"], ) logger.reason(f"TIMING: LLM call: {time.monotonic() - t_llm:.2f}s", {}) diff --git a/backend/src/plugins/translate/preview_executor.py b/backend/src/plugins/translate/preview_executor.py index 05931159..358e38e7 100644 --- a/backend/src/plugins/translate/preview_executor.py +++ b/backend/src/plugins/translate/preview_executor.py @@ -18,7 +18,7 @@ from ...core.config_manager import ConfigManager from ...core.logger import belief_scope, logger from ...core.superset_client import SupersetClient from ...models.translate import TranslationJob -from .preview_llm_client import LLMClient +from ._llm_async_http import call_openai_compatible from .preview_resolve_provider import resolve_provider_model as _resolve_provider_model from .preview_response_parser import ( compute_config_hash as _compute_config_hash, @@ -38,7 +38,7 @@ class PreviewExecutor: # region fetch_sample_rows [TYPE Function] # @PURPOSE: Fetch sample rows from Superset dataset for preview. # @SIDE_EFFECT Calls Superset chart data endpoint. - def fetch_sample_rows( + async def fetch_sample_rows( self, job: TranslationJob, sample_size: int = 10, @@ -55,7 +55,7 @@ class PreviewExecutor: raise ValueError("No Superset environments configured") client = SupersetClient(env_config) - dataset_detail = client.get_dataset_detail(int(job.source_datasource_id)) + dataset_detail = await client.get_dataset_detail(int(job.source_datasource_id)) query_context = client.build_dataset_preview_query_context( dataset_id=int(job.source_datasource_id), dataset_record=dataset_detail, template_params={}, effective_filters=[], @@ -71,7 +71,7 @@ class PreviewExecutor: form_data.pop("query_mode", None) try: - response = client.network.request( + response = await client.network.request( method="POST", endpoint="/api/v1/chart/data", data=json.dumps(query_context), headers={"Content-Type": "application/json"}, @@ -85,7 +85,7 @@ class PreviewExecutor: # region call_llm [TYPE Function] # @PURPOSE: Call the configured LLM provider with a prompt. # @SIDE_EFFECT Makes HTTP call to LLM provider. - def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str: + async def call_llm(self, job: TranslationJob, prompt: str, max_tokens: int = 8192) -> str: with belief_scope("PreviewExecutor.call_llm"): if not job.provider_id: raise ValueError("Job has no LLM provider configured") @@ -108,7 +108,7 @@ class PreviewExecutor: max_attempts = 2 for attempt in range(max_attempts): try: - response_text = LLMClient.call_openai_compatible( + content_text, _ = await call_openai_compatible( base_url=provider.base_url, api_key=api_key, model=model, prompt=prompt, provider_type=provider_type, max_tokens=max_tokens * (attempt + 1), @@ -121,7 +121,7 @@ class PreviewExecutor: continue raise - return response_text + return content_text # endregion call_llm # region delegation_helpers [TYPE Function] diff --git a/backend/src/plugins/translate/superset_executor.py b/backend/src/plugins/translate/superset_executor.py index 6b18ec73..a5f16575 100644 --- a/backend/src/plugins/translate/superset_executor.py +++ b/backend/src/plugins/translate/superset_executor.py @@ -9,8 +9,8 @@ # @RATIONALE Direct SQL Lab API submission provides audit trail and execution monitoring within Superset. # @REJECTED Direct database connection bypass — would skip Superset's SQL Lab audit and RBAC. +import asyncio import json -import time from typing import Any import uuid @@ -146,7 +146,7 @@ class SupersetSqlLabExecutor: # @PRE sql is valid SQL string. database_id is a valid Superset DB ID. # @POST Returns execution result dict with query_id and status. # @SIDE_EFFECT Makes HTTP POST to /api/v1/sqllab/execute/. - def execute_sql( + async def execute_sql( self, sql: str, database_id: int | None = None, @@ -180,7 +180,7 @@ class SupersetSqlLabExecutor: # fetch() calls in Superset SQL Lab. endpoint = "/api/v1/sqllab/execute/" try: - response = client.network.request( + response = await client.network.request( method="POST", endpoint=endpoint, data=json.dumps(payload), @@ -248,7 +248,7 @@ class SupersetSqlLabExecutor: # @PRE query_id is a valid Superset query ID. # @POST Returns final execution status dict. # @SIDE_EFFECT Makes HTTP GET requests to Superset. - def poll_execution_status( + async def poll_execution_status( self, query_id: str, max_polls: int = 60, @@ -265,7 +265,7 @@ class SupersetSqlLabExecutor: for attempt in range(max_polls): try: - response = client.network.request( + response = await client.network.request( method="GET", endpoint=f"/api/v1/query/{query_id}", ) @@ -308,11 +308,11 @@ class SupersetSqlLabExecutor: } if state in ("pending", "running", "started"): - time.sleep(poll_interval_seconds) + await asyncio.sleep(poll_interval_seconds) continue # Unknown state — treat as still running - time.sleep(poll_interval_seconds) + await asyncio.sleep(poll_interval_seconds) except Exception as e: logger.explore("Polling error, retrying", { @@ -320,7 +320,7 @@ class SupersetSqlLabExecutor: "attempt": attempt + 1, "error": str(e), }) - time.sleep(poll_interval_seconds) + await asyncio.sleep(poll_interval_seconds) continue # Timeout @@ -341,7 +341,7 @@ class SupersetSqlLabExecutor: # @PRE sql is valid SQL. # @POST Returns final execution result. # @SIDE_EFFECT Makes HTTP calls to Superset API. - def execute_and_poll( + async def execute_and_poll( self, sql: str, database_id: int | None = None, @@ -349,7 +349,7 @@ class SupersetSqlLabExecutor: poll_interval_seconds: float = 2.0, ) -> dict[str, Any]: with belief_scope("SupersetSqlLabExecutor.execute_and_poll"): - exec_result = self.execute_sql( + exec_result = await self.execute_sql( sql=sql, database_id=database_id, run_async=False, # Sync mode — Superset returns result directly @@ -384,7 +384,7 @@ class SupersetSqlLabExecutor: "query_id": None, } - return self.poll_execution_status( + return await self.poll_execution_status( query_id=str(query_id), max_polls=max_polls, poll_interval_seconds=poll_interval_seconds, @@ -396,11 +396,11 @@ class SupersetSqlLabExecutor: # @PRE query_id is a valid Superset query ID. # @POST Returns query results if available. # @SIDE_EFFECT Makes HTTP GET to Superset. - def get_query_results(self, query_id: str) -> dict[str, Any]: + async def get_query_results(self, query_id: str) -> dict[str, Any]: with belief_scope("SupersetSqlLabExecutor.get_query_results"): client = self._get_client() try: - response = client.network.request( + response = await client.network.request( method="GET", endpoint=f"/api/v1/query/{query_id}/results", )