From 3f6d71ecbcd60498f72d2728ca1826628f1e198b Mon Sep 17 00:00:00 2001 From: KoalaTrapLord <265133236+KoalaTrapLord@users.noreply.github.com> Date: Fri, 27 Mar 2026 02:36:22 -0400 Subject: [PATCH] fix(coach): switch all Anthropic calls to streaming to prevent 10-min timeout error (closes #109) - anthropic_client.py: all 4 methods (chat_structured, chat_plain, achat_structured, achat_plain) now use .messages.stream() context manager - coach_service.py: inline aclient.messages.create() replaced with async streaming via aclient.messages.stream() - _parse_response() updated to accept pre-extracted text string (since stream gives us text directly) - check_status() left as non-streaming (max_tokens=10, safe from the 10min rule) --- coach/clients/anthropic_client.py | 83 +++++++++++++++---------------- coach/coach_service.py | 61 ++++++++--------------- 2 files changed, 60 insertions(+), 84 deletions(-) diff --git a/coach/clients/anthropic_client.py b/coach/clients/anthropic_client.py index ba7be65..19d6567 100644 --- a/coach/clients/anthropic_client.py +++ b/coach/clients/anthropic_client.py @@ -4,6 +4,9 @@ Commander AI Lab — Anthropic Claude Structured Output Client Uses the anthropic SDK to call Claude's messages endpoint with structured JSON output enforced via system prompt. Supports both sync and native async operation. + +All substantive calls use .messages.stream() to avoid the SDK's +10-minute non-streaming timeout restriction on large max_tokens. """ import json import logging @@ -53,6 +56,8 @@ class AnthropicClient: Anthropic Claude API client. Drop-in replacement for PerplexityClient. + All substantive calls use .messages.stream() so the SDK never + enforces the 10-minute non-streaming timeout. Usage: client = AnthropicClient(api_key="sk-ant-...") @@ -101,37 +106,24 @@ class AnthropicClient: ) -> AnthropicResponse: """ Send a chat request expecting JSON output. - - The system_prompt should instruct the model to return JSON matching - the expected schema. Claude Opus follows JSON instructions reliably. - - Args: - system_prompt: System message (should instruct JSON output) - user_prompt: User message - json_schema: (optional) Kept for call-site compatibility; appended - as a schema hint in the system prompt when provided - schema_name: (optional) Label used in the schema hint - model: Override model - temperature: Sampling temperature (0.0-1.0) - max_tokens: Max output tokens - - Returns: - AnthropicResponse with parsed_json extracted from model output + Uses streaming to avoid the Anthropic SDK 10-minute timeout restriction. """ use_model = model or self.model effective_system = self._build_system(system_prompt, json_schema, schema_name) try: - response = self._client.messages.create( + with self._client.messages.stream( model=use_model, system=effective_system, messages=[{"role": "user", "content": user_prompt}], temperature=temperature, max_tokens=max_tokens, - ) - return self._parse_response(response, use_model) + ) as stream: + content = stream.get_final_text() + final_msg = stream.get_final_message() + return self._parse_streamed(content, final_msg, use_model) except Exception as e: - logger.error("[anthropic] API call failed: %s", e) + logger.error("[anthropic] chat_structured failed: %s", e) raise def chat_plain( @@ -144,21 +136,23 @@ class AnthropicClient: ) -> AnthropicResponse: """ Send a plain chat request without strict structured output. - Used for substitution fallback queries. + Uses streaming to avoid the Anthropic SDK 10-minute timeout restriction. """ use_model = model or self.model try: - response = self._client.messages.create( + with self._client.messages.stream( model=use_model, system=system_prompt, messages=[{"role": "user", "content": user_prompt}], temperature=temperature, max_tokens=max_tokens, - ) - return self._parse_response(response, use_model) + ) as stream: + content = stream.get_final_text() + final_msg = stream.get_final_message() + return self._parse_streamed(content, final_msg, use_model) except Exception as e: - logger.error("[anthropic] Plain chat failed: %s", e) + logger.error("[anthropic] chat_plain failed: %s", e) raise async def achat_structured( @@ -171,21 +165,23 @@ class AnthropicClient: temperature: float = 0.2, max_tokens: int = 8192, ) -> AnthropicResponse: - """Native async structured chat — no run_in_executor needed.""" + """Native async structured chat using streaming.""" use_model = model or self.model effective_system = self._build_system(system_prompt, json_schema, schema_name) try: - response = await self._async_client.messages.create( + async with self._async_client.messages.stream( model=use_model, system=effective_system, messages=[{"role": "user", "content": user_prompt}], temperature=temperature, max_tokens=max_tokens, - ) - return self._parse_response(response, use_model) + ) as stream: + content = await stream.get_final_text() + final_msg = await stream.get_final_message() + return self._parse_streamed(content, final_msg, use_model) except Exception as e: - logger.error("[anthropic] Async structured chat failed: %s", e) + logger.error("[anthropic] achat_structured failed: %s", e) raise async def achat_plain( @@ -196,24 +192,29 @@ class AnthropicClient: temperature: float = 0.2, max_tokens: int = 4096, ) -> AnthropicResponse: - """Native async plain chat.""" + """Native async plain chat using streaming.""" use_model = model or self.model try: - response = await self._async_client.messages.create( + async with self._async_client.messages.stream( model=use_model, system=system_prompt, messages=[{"role": "user", "content": user_prompt}], temperature=temperature, max_tokens=max_tokens, - ) - return self._parse_response(response, use_model) + ) as stream: + content = await stream.get_final_text() + final_msg = await stream.get_final_message() + return self._parse_streamed(content, final_msg, use_model) except Exception as e: - logger.error("[anthropic] Async plain chat failed: %s", e) + logger.error("[anthropic] achat_plain failed: %s", e) raise def check_status(self) -> dict: - """Quick health check — try a minimal API call.""" + """ + Quick health check — tiny max_tokens=10 call, safe to use non-streaming + since it will never approach the 10-minute threshold. + """ try: response = self._client.messages.create( model=self.model, @@ -249,13 +250,9 @@ class AnthropicClient: f"{schema_str}" ) - def _parse_response(self, response, use_model: str) -> AnthropicResponse: - """Extract content and attempt JSON parsing from an Anthropic response.""" - content = "" - if response.content: - content = response.content[0].text or "" - - usage = response.usage + def _parse_streamed(self, content: str, final_msg, use_model: str) -> AnthropicResponse: + """Build an AnthropicResponse from streamed text + final message metadata.""" + usage = getattr(final_msg, "usage", None) prompt_tokens = usage.input_tokens if usage else 0 completion_tokens = usage.output_tokens if usage else 0 diff --git a/coach/coach_service.py b/coach/coach_service.py index c2c4288..5e93001 100644 --- a/coach/coach_service.py +++ b/coach/coach_service.py @@ -65,10 +65,8 @@ class CoachService: def load_deck_report(self, deck_id: str) -> Optional[DeckReport]: """Load a deck report from disk.""" - # Try exact filename first, then case-insensitive search report_path = DECK_REPORTS_DIR / f"{deck_id}.json" if not report_path.exists(): - # Try case-insensitive match for f in DECK_REPORTS_DIR.glob("*.json"): if re.sub(r'[^a-z0-9]', '', f.stem.lower()) == re.sub(r'[^a-z0-9]', '', deck_id.lower()): report_path = f @@ -96,12 +94,10 @@ class CoachService: def save_session(self, session: CoachSession) -> Path: """Persist a coaching session to disk and SQLite.""" - # 1. Write JSON file (backward compat) COACH_SESSIONS_DIR.mkdir(parents=True, exist_ok=True) path = COACH_SESSIONS_DIR / f"{session.sessionId}.json" with open(path, "w") as f: f.write(session.model_dump_json(indent=2)) - # 2. Write to SQLite if _DB_AVAILABLE: try: _db_save_session({ @@ -124,10 +120,8 @@ class CoachService: logger.warning("Failed to persist session to SQLite: %s", e) return path - def load_session(self, session_id: str) -> Optional[CoachSession]: """Load a coaching session from SQLite, falling back to disk.""" - # Try SQLite first if _DB_AVAILABLE: try: row = _db_load_session(session_id) @@ -150,7 +144,6 @@ class CoachService: ) except Exception as e: logger.warning("SQLite load failed for %s: %s", session_id, e) - # Fallback to disk path = COACH_SESSIONS_DIR / f"{session_id}.json" if not path.exists(): return None @@ -162,10 +155,8 @@ class CoachService: logger.error("Failed to load session %s: %s", session_id, e) return None - def list_sessions(self, deck_id: str = None) -> List[dict]: """List coaching sessions from SQLite, falling back to disk.""" - # Try SQLite first if _DB_AVAILABLE: try: rows = _db_list_sessions(deck_id) @@ -173,7 +164,6 @@ class CoachService: return rows except Exception as e: logger.warning("SQLite list_sessions failed: %s", e) - # Fallback to disk if not COACH_SESSIONS_DIR.exists(): return [] sessions = [] @@ -195,20 +185,21 @@ class CoachService: continue return sessions - # ── Main Coaching Pipeline ───────────────────────────────── - async def run_coaching_session(self, deck_id: str, - goals: Optional[CoachGoals] = None, - fallback_report: Optional[DeckReport] = None, - ) -> CoachSession: + async def run_coaching_session( + self, + deck_id: str, + goals: Optional[CoachGoals] = None, + fallback_report: Optional[DeckReport] = None, + ) -> CoachSession: """ Full coaching pipeline: 1. Load deck report (or use fallback from DB card list) 2. Identify underperformers 3. Find replacement candidates via embeddings 4. Build prompt - 5. Call LLM + 5. Call LLM (streaming for Anthropic to avoid 10-min SDK timeout) 6. Parse response 7. Persist session """ @@ -224,7 +215,7 @@ class CoachService: except Exception as e: logger.warning("Embeddings auto-load failed: %s", e) - # 1. Load deck report — try simulation report first, then fallback + # 1. Load deck report report = self.load_deck_report(deck_id) if report is None and fallback_report is not None: report = fallback_report @@ -237,20 +228,15 @@ class CoachService: # 3. Find replacement candidates for underperformers candidates: Dict[str, List[dict]] = {} - underperformers = report.underperformers[:MAX_UNDERPERFORMERS] if not underperformers: - # If no explicit underperformers, pick lowest impact cards sorted_cards = sorted(report.cards, key=lambda c: c.impactScore) underperformers = [ c.name for c in sorted_cards[:MAX_UNDERPERFORMERS] if c.impactScore < UNDERPERFORMER_IMPACT_THRESHOLD ] - - # If still no underperformers and we have cards, pick bottom N by impact if not underperformers and report.cards: sorted_cards = sorted(report.cards, key=lambda c: c.impactScore) - # Take the bottom 5 non-land cards as candidates for cuts underperformers = [ c.name for c in sorted_cards[:5] if not any(t in (c.tags or []) for t in []) @@ -273,29 +259,33 @@ class CoachService: system_prompt = build_system_prompt(report, goals) user_prompt = build_user_prompt(report, candidates) - # 5. Call LLM (Anthropic or local Ollama) + # 5. Call LLM logger.info("Calling LLM for deck: %s (provider=%s)", deck_id, COACH_PROVIDER) if COACH_PROVIDER == "anthropic": - import anthropic + import anthropic as _anthropic anthropic_key = ANTHROPIC_API_KEY if not anthropic_key: raise ConnectionError("Anthropic API key not configured. Set ANTHROPIC_API_KEY env var.") - aclient = anthropic.AsyncAnthropic(api_key=anthropic_key) - resp = await aclient.messages.create( + + # Use streaming to avoid the Anthropic SDK 10-minute non-streaming timeout. + aclient = _anthropic.AsyncAnthropic(api_key=anthropic_key) + async with aclient.messages.stream( model=ANTHROPIC_MODEL, system=system_prompt, messages=[{"role": "user", "content": user_prompt}], max_tokens=DEFAULT_MAX_TOKENS, temperature=0.7, - ) - content = resp.content[0].text if resp.content else "" - usage = resp.usage + ) as stream: + content = await stream.get_final_text() + final_msg = await stream.get_final_message() + + usage = getattr(final_msg, "usage", None) # Build an LLMResponse-compatible object from types import SimpleNamespace llm_response = SimpleNamespace( content=content, - model=resp.model, + model=final_msg.model, prompt_tokens=usage.input_tokens if usage else 0, completion_tokens=usage.output_tokens if usage else 0, parsed_json=None, @@ -305,7 +295,6 @@ class CoachService: cleaned = re.sub(r'\s*```$', '', cleaned.strip()) llm_response.parsed_json = json.loads(cleaned) except (json.JSONDecodeError, ValueError): - # Try to find JSON object in the text match = re.search(r'\{[\s\S]*\}', content) if match: try: @@ -317,7 +306,6 @@ class CoachService: # 6. Parse response into CoachSession session_id = f"sess-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" - session = CoachSession( sessionId=session_id, deckId=deck_id, @@ -331,14 +319,12 @@ class CoachService: if llm_response.parsed_json: self._populate_session_from_json(session, llm_response.parsed_json) else: - # Fallback: store raw text session.rawTextExplanation = llm_response.content session.summary = "LLM returned non-JSON response. See raw explanation." # 7. Persist self.save_session(session) logger.info("Coaching session saved: %s", session_id) - return session def _populate_session_from_json(self, session: CoachSession, data: dict): @@ -347,11 +333,9 @@ class CoachService: session.rawTextExplanation = data.get("rawTextExplanation", "") session.manaBaseAdvice = data.get("manaBaseAdvice") - # Heuristic hints hints = data.get("heuristicHints", []) session.heuristicHints = hints if isinstance(hints, list) else [str(hints)] - # Suggested cuts for cut_data in data.get("suggestedCuts", []): if isinstance(cut_data, dict): session.suggestedCuts.append(SuggestedCut( @@ -360,7 +344,6 @@ class CoachService: replacementOptions=cut_data.get("replacementOptions", []), )) - # Suggested adds for add_data in data.get("suggestedAdds", []): if isinstance(add_data, dict): session.suggestedAdds.append(SuggestedAdd( @@ -376,7 +359,6 @@ class CoachService: """Check the health of all coach subsystems.""" status = CoachStatus() - # Check LLM / Anthropic provider if COACH_PROVIDER == "anthropic": anthropic_key = ANTHROPIC_API_KEY if anthropic_key: @@ -396,11 +378,8 @@ class CoachService: if not status.llmConnected: status.error = llm_status.get("error", "Local LLM not reachable") - # Check embeddings status.embeddingsLoaded = self.embeddings.loaded status.embeddingCards = self.embeddings.card_count - - # Check deck reports status.deckReportsAvailable = len(self.list_deck_reports()) return status