()
| 39 | def _load_adapter_module(): |
| 40 | from app.services import tenvision_adapter |
| 41 | |
| 42 | return tenvision_adapter |
| 43 | |
| 44 | |
| 45 | def _worker_initializer() -> None: |
| 46 | _clear_proxy_env() |
| 47 | _load_adapter_module().get_engine() |
| 48 | |
| 49 | |
| 50 | def _warmup_worker(index: int) -> dict[str, Any]: |
| 51 | _clear_proxy_env() |
| 52 | _load_adapter_module().get_engine() |
| 53 | time.sleep(0.1) |
| 54 | return {"index": index, "pid": os.getpid()} |
| 55 | |
| 56 | |
| 57 | def _worker_analyze(data: bytes, prompt_text: str, include_debug: bool) -> dict[str, Any]: |
no test coverage detected