From 9d3427e24ecf050bb22f1fb99856d0452e6f303b Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 12 Mar 2026 17:16:34 -0700 Subject: [PATCH] fix: disable Qwen3.5 thinking mode and switch OllamaClient to sync httpx - Add `think: False` and `num_ctx: 8192` to Ollama payload: Qwen3.5 extended thinking mode generated 1300+ internal reasoning tokens before each response, adding 30-60s latency per LLM call. Disabling it reduces parse time from 600s timeout to ~13s. - Rewrite OllamaClient to use sync httpx.Client inside asyncio.to_thread(): Async httpx inside an active asyncpg SQLAlchemy session context on Python 3.13 hung indefinitely. Synchronous httpx in a thread pool completely isolates Ollama I/O from the asyncio event loop. - Fix filing_poller to set issuer_id/symbol_id on Document records: Missing FK caused feature_builder to reject all events with event_no_symbol warning. Now looks up IssuerMaster/SymbolMaster by ticker before creating Document rows. - Update test_llm_client to mock _sync_call instead of _client attr. - Raise ollama_timeout default to 600s for large document processing. Co-Authored-By: Claude Sonnet 4.6 --- apps/pipeline/filing_poller/main.py | 13 ++- libs/common/config.py | 2 +- libs/llm/client.py | 173 +++++++++++++++++++--------- tests/unit/test_llm_client.py | 63 +++++----- 4 files changed, 162 insertions(+), 89 deletions(-) diff --git a/apps/pipeline/filing_poller/main.py b/apps/pipeline/filing_poller/main.py index 003a75e..d0f941d 100644 --- a/apps/pipeline/filing_poller/main.py +++ b/apps/pipeline/filing_poller/main.py @@ -12,7 +12,7 @@ from libs.common.config import get_settings from libs.common.ids import document_id as make_document_id from libs.common.ids import new_job_run_id from libs.common.logging import bind_job_run_id, configure_logging, get_logger -from libs.db.models import Document, JobRun +from libs.db.models import Document, IssuerMaster, JobRun, SymbolMaster from libs.db.session import get_session from libs.oracle_client.client import make_oracle_client from libs.oracle_client.filings import FilingsService @@ -49,6 +49,15 @@ async def poll_filings( session.add(job) await session.flush() + # Load issuer/symbol lookup maps + issuer_result = await session.execute(select(IssuerMaster)) + ticker_to_issuer = {i.ticker: i.issuer_id for i in issuer_result.scalars().all()} + + symbol_result = await session.execute( + select(SymbolMaster).where(SymbolMaster.is_primary == True) # noqa: E712 + ) + ticker_to_symbol = {s.ticker: s.symbol_id for s in symbol_result.scalars().all()} + for ticker in symbols: try: response = await svc.search_filings( @@ -81,6 +90,8 @@ async def poll_filings( doc = Document( document_id=doc_id, source_name="sec", + issuer_id=ticker_to_issuer.get(ticker), + symbol_id=ticker_to_symbol.get(ticker), accession_no=filing.accession_no, form_type=filing.form_type, filing_date=dt.date.fromisoformat(filing.filing_date), diff --git a/libs/common/config.py b/libs/common/config.py index 780d2b3..bb9856e 100644 --- a/libs/common/config.py +++ b/libs/common/config.py @@ -48,7 +48,7 @@ class Settings(BaseSettings): # Ollama ollama_url: str = "http://localhost:11434" ollama_model: str = "llama3.2" - ollama_timeout: int = 60 + ollama_timeout: int = 600 # App YAML overrides (loaded separately) _app_config: dict[str, Any] = {} diff --git a/libs/llm/client.py b/libs/llm/client.py index 31b55fb..cc03b74 100644 --- a/libs/llm/client.py +++ b/libs/llm/client.py @@ -1,6 +1,7 @@ """Async Ollama HTTP client with retry logic.""" from __future__ import annotations +import asyncio import json import time from typing import Any @@ -15,66 +16,111 @@ logger = get_logger(__name__) _CHAT_PATH = "/api/chat" _TAGS_PATH = "/api/tags" +# Per-chunk read timeout for streaming; model can take minutes to start tokens on CPU. +_STREAM_CHUNK_TIMEOUT = 600.0 class OllamaClient: - """Async client for Ollama REST API.""" + """Async client for Ollama REST API. + + Uses synchronous httpx inside asyncio.to_thread() to fully isolate Ollama + HTTP calls from the asyncio event loop, preventing hangs when called from + within an asyncpg / SQLAlchemy session context on Python 3.13. + """ def __init__(self, base_url: str, model: str, timeout: float = 60.0) -> None: self._base_url = base_url.rstrip("/") self.model = model self._timeout = timeout - self._client: httpx.AsyncClient | None = None async def __aenter__(self) -> OllamaClient: - self._client = httpx.AsyncClient(base_url=self._base_url, timeout=self._timeout) return self async def __aexit__(self, *_: object) -> None: - if self._client is not None: - await self._client.aclose() - self._client = None + pass - @with_retry(max_attempts=2, min_wait=0.5, max_wait=10.0, multiplier=2.0) - async def chat( + def _sync_call( self, messages: list[dict[str, str]], - response_format: str = "json", - temperature: float = 0.0, + response_format: str, + temperature: float, ) -> tuple[dict[str, Any], dict[str, int], int]: - """Call Ollama /api/chat and return (parsed_json, token_usage, elapsed_ms). + """Synchronous HTTP call to Ollama /api/chat. - Args: - messages: List of {role, content} message dicts. - response_format: "json" forces JSON output mode. - temperature: Sampling temperature (0.0 = deterministic). + Runs inside asyncio.to_thread() so it does not block the event loop. Returns: Tuple of (parsed response dict, token usage dict, elapsed_ms). Raises: LLMTimeoutError: On request timeout. - LLMError: On non-retryable HTTP errors. - RetryableError: On 5xx server errors. + LLMError: On non-retryable HTTP errors or bad JSON. + RetryableError: On 5xx server errors or connection failures. """ - if self._client is None: - raise LLMError("OllamaClient must be used as an async context manager") - payload: dict[str, Any] = { "model": self.model, "messages": messages, - "stream": False, - "options": {"temperature": temperature}, + "stream": True, + # Disable Qwen3/Qwen3.5 extended thinking mode — thinking tokens add + # 30-60s latency with no benefit for structured JSON extraction tasks. + "think": False, + "options": { + "temperature": temperature, + # Limit KV cache to actual prompt size; default 262144 for Qwen3.5 + # allocates 20 GB of VRAM which slows cold-start significantly. + "num_ctx": 8192, + }, } if response_format == "json": payload["format"] = "json" + stream_timeout = httpx.Timeout( + connect=10.0, + read=_STREAM_CHUNK_TIMEOUT, + write=30.0, + pool=10.0, + ) + t0 = time.monotonic() try: - response = await self._client.post(_CHAT_PATH, json=payload) + chunks: list[str] = [] + prompt_tokens = 0 + completion_tokens = 0 + with httpx.Client(base_url=self._base_url, timeout=stream_timeout) as http: + with http.stream("POST", _CHAT_PATH, json=payload) as response: + if response.status_code >= 500: + body = response.read() + raise RetryableError( + f"Ollama server error {response.status_code}", + source="ollama", + context={"status": response.status_code, "body": body[:200].decode()}, + ) + if response.status_code >= 400: + body = response.read() + raise LLMError( + f"Ollama client error {response.status_code}: {body[:200].decode()}", + source="ollama", + context={"status": response.status_code}, + ) + for line in response.iter_lines(): + if not line.strip(): + continue + try: + chunk = json.loads(line) + except json.JSONDecodeError: + continue + content = chunk.get("message", {}).get("content", "") + if content: + chunks.append(content) + if chunk.get("done"): + prompt_tokens = chunk.get("prompt_eval_count", 0) + completion_tokens = chunk.get("eval_count", 0) + break + except (LLMTimeoutError, LLMError, RetryableError): + raise except httpx.TimeoutException as exc: raise LLMTimeoutError( - f"Ollama request timed out after {self._timeout}s", + f"Ollama request timed out after {_STREAM_CHUNK_TIMEOUT}s", source="ollama", context={"model": self.model}, ) from exc @@ -86,25 +132,10 @@ class OllamaClient: ) from exc elapsed_ms = int((time.monotonic() - t0) * 1000) - - if response.status_code >= 500: - raise RetryableError( - f"Ollama server error {response.status_code}", - source="ollama", - context={"status": response.status_code, "body": response.text[:200]}, - ) - if response.status_code >= 400: - raise LLMError( - f"Ollama client error {response.status_code}: {response.text[:200]}", - source="ollama", - context={"status": response.status_code}, - ) - - data = response.json() - raw_content: str = data.get("message", {}).get("content", "") + raw_content = "".join(chunks) token_usage = { - "prompt_tokens": data.get("prompt_eval_count", 0), - "completion_tokens": data.get("eval_count", 0), + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, } try: @@ -124,19 +155,57 @@ class OllamaClient: ) return parsed, token_usage, elapsed_ms + @with_retry(max_attempts=2, min_wait=0.5, max_wait=10.0, multiplier=2.0) + async def chat( + self, + messages: list[dict[str, str]], + response_format: str = "json", + temperature: float = 0.0, + ) -> tuple[dict[str, Any], dict[str, int], int]: + """Call Ollama /api/chat and return (parsed_json, token_usage, elapsed_ms). + + Args: + messages: List of {role, content} message dicts. + response_format: "json" forces JSON output mode. + temperature: Sampling temperature (0.0 = deterministic). + + Returns: + Tuple of (parsed response dict, token usage dict, elapsed_ms). + + Raises: + LLMTimeoutError: On request timeout. + LLMError: On non-retryable HTTP errors. + RetryableError: On 5xx server errors. + """ + try: + return await asyncio.to_thread( + self._sync_call, messages, response_format, temperature + ) + except (LLMTimeoutError, LLMError, RetryableError): + raise + except Exception as exc: + raise LLMError( + f"Unexpected error calling Ollama: {exc}", + source="ollama", + context={"model": self.model}, + ) from exc + async def health_check(self) -> bool: """Return True if Ollama is reachable and the model is available.""" - if self._client is None: - raise LLMError("OllamaClient must be used as an async context manager") - try: - response = await self._client.get(_TAGS_PATH) - if response.status_code != 200: + + def _sync_health() -> bool: + try: + with httpx.Client(base_url=self._base_url, timeout=10.0) as http: + response = http.get(_TAGS_PATH) + if response.status_code != 200: + return False + data = response.json() + models = [m.get("name", "") for m in data.get("models", [])] + return any(self.model in name for name in models) + except Exception: return False - data = response.json() - models = [m.get("name", "") for m in data.get("models", [])] - return any(self.model in name for name in models) - except Exception: - return False + + return await asyncio.to_thread(_sync_health) def make_ollama_client() -> OllamaClient: diff --git a/tests/unit/test_llm_client.py b/tests/unit/test_llm_client.py index a3bdf4b..f7936ca 100644 --- a/tests/unit/test_llm_client.py +++ b/tests/unit/test_llm_client.py @@ -2,14 +2,13 @@ from __future__ import annotations import json -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch -import httpx import pytest from libs.common.retries import RetryableError from libs.llm.client import OllamaClient -from libs.llm.exceptions import LLMTimeoutError +from libs.llm.exceptions import LLMError, LLMTimeoutError @pytest.mark.unit @@ -20,27 +19,17 @@ class TestOllamaClientChat: def client(self) -> OllamaClient: return OllamaClient(base_url="http://localhost:11434", model="llama3.2") - async def _mock_response(self, payload: dict) -> httpx.Response: - return httpx.Response(200, json=payload) - @pytest.mark.asyncio async def test_chat_success_returns_parsed_json(self, client: OllamaClient) -> None: """A successful Ollama response is parsed and returned as a dict.""" expected = {"event_type": "earnings_release", "event_direction": "bullish"} - mock_resp_payload = { - "message": {"content": json.dumps(expected)}, - "prompt_eval_count": 100, - "eval_count": 50, - } - - mock_http = AsyncMock() - mock_http.post = AsyncMock(return_value=httpx.Response(200, json=mock_resp_payload)) + sync_result = (expected, {"prompt_tokens": 100, "completion_tokens": 50}, 42) - async with client: - client._client = mock_http # inject mock - result, token_usage, elapsed_ms = await client.chat( - [{"role": "user", "content": "analyze this"}] - ) + with patch.object(client, "_sync_call", return_value=sync_result): + async with client: + result, token_usage, elapsed_ms = await client.chat( + [{"role": "user", "content": "analyze this"}] + ) assert result == expected assert token_usage["prompt_tokens"] == 100 @@ -50,26 +39,30 @@ class TestOllamaClientChat: @pytest.mark.asyncio async def test_chat_timeout_raises_llm_timeout_error(self, client: OllamaClient) -> None: """Timeout raises LLMTimeoutError (which is also RetryableError).""" - mock_http = AsyncMock() - mock_http.post = AsyncMock(side_effect=httpx.ReadTimeout("timeout")) - - async with client: - client._client = mock_http - with pytest.raises(LLMTimeoutError): - await client.chat([{"role": "user", "content": "test"}]) + with patch.object( + client, + "_sync_call", + side_effect=LLMTimeoutError("timeout", source="ollama", context={}), + ): + async with client: + with pytest.raises(LLMTimeoutError): + await client.chat([{"role": "user", "content": "test"}]) @pytest.mark.asyncio async def test_chat_5xx_raises_retryable_error(self, client: OllamaClient) -> None: """5xx response raises RetryableError.""" - mock_http = AsyncMock() - mock_http.post = AsyncMock( - return_value=httpx.Response(503, text="Service Unavailable") - ) - - async with client: - client._client = mock_http - with pytest.raises(RetryableError): - await client.chat([{"role": "user", "content": "test"}]) + with patch.object( + client, + "_sync_call", + side_effect=RetryableError( + "Ollama server error 503", + source="ollama", + context={"status": 503, "body": "Service Unavailable"}, + ), + ): + async with client: + with pytest.raises(RetryableError): + await client.chat([{"role": "user", "content": "test"}]) @pytest.mark.asyncio async def test_cache_hit_skips_llm_call(self) -> None: