fez5587/commander-ai-lab
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(coach): switch all Anthropic calls to streaming to prevent 10-min timeout error (closes #109)

Browse Source 
- 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)
This commit is contained in:
KoalaTrapLord committed 2026-03-27 02:36:22 -04:00
1 parent 4246555228
commit 3f6d71ecbc
2 files changed
+60 -84
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

coach/clients/anthropic_client.py
+40 -43
View File
@@ -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
coach/coach_service.py
+20 -41
View File
@@ -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