Spaces:
Sleeping
Sleeping
File size: 3,596 Bytes
331f4c6 6848579 331f4c6 a06c9f9 331f4c6 a06c9f9 331f4c6 a06c9f9 331f4c6 a06c9f9 331f4c6 a06c9f9 331f4c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """
Nimbus Bank Triage — LLM Client Wrappers
Centralized model configuration for Claude Haiku (fast tasks)
and Claude Sonnet (quality drafting), with retry logic.
"""
import os
from functools import lru_cache
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from tenacity import retry, stop_after_attempt, wait_exponential
# Load env vars
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
load_dotenv(os.path.join(ROOT_DIR, ".env"))
# ── Model IDs ────────────────────────────────────────────────
FAST_MODEL = os.environ.get("FAST_MODEL", "claude-haiku-4-5-20251001")
# Default the drafter to the same known-working model unless overridden by env.
DRAFTER_MODEL = os.environ.get("DRAFTER_MODEL", "claude-haiku-4-5-20251001")
def _get_anthropic_api_key() -> str:
"""Return a non-empty Anthropic API key or raise a clear config error."""
api_key = os.environ.get("ANTHROPIC_API_KEY", "").strip()
if not api_key:
raise RuntimeError(
"ANTHROPIC_API_KEY is not configured. Add it to your environment or Hugging Face Space secrets, then restart the app."
)
return api_key
def anthropic_api_key_configured() -> bool:
"""Whether a non-empty Anthropic API key is currently configured."""
return bool(os.environ.get("ANTHROPIC_API_KEY", "").strip())
@lru_cache(maxsize=4)
def _build_llm(model: str, temperature: float, max_tokens: int, api_key: str) -> ChatAnthropic:
"""Build and cache Anthropic chat clients by model settings and API key."""
return ChatAnthropic(
model=model,
temperature=temperature,
max_tokens=max_tokens,
api_key=api_key,
)
@lru_cache(maxsize=1)
def get_fast_llm() -> ChatAnthropic:
"""
Claude Haiku — used for classification, compliance checks,
and injection detection. Low latency, low cost, deterministic.
"""
return _build_llm(
model=FAST_MODEL,
temperature=0.0,
max_tokens=1024,
api_key=_get_anthropic_api_key(),
)
@lru_cache(maxsize=1)
def get_drafter_llm() -> ChatAnthropic:
"""
Claude Sonnet — used only by the Response Drafter.
Higher quality for nuanced, empathetic customer responses.
"""
return _build_llm(
model=DRAFTER_MODEL,
temperature=0.5,
max_tokens=2048,
api_key=_get_anthropic_api_key(),
)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
reraise=True,
)
def invoke_with_retry(llm: ChatAnthropic, messages: list) -> str:
"""
Invoke an LLM with automatic retry on transient failures.
Exponential backoff: 1s, 2s, 4s. Max 3 attempts.
Returns the string content of the response.
"""
response = llm.invoke(messages)
return response.content
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
reraise=True,
)
def invoke_structured_with_retry(
llm: ChatAnthropic,
messages: list,
schema: type,
) -> dict:
"""
Invoke an LLM with structured output (Pydantic schema) and retry.
Uses LangChain's with_structured_output for type-safe responses.
Returns a dict matching the schema.
"""
structured_llm = llm.with_structured_output(schema)
response = structured_llm.invoke(messages)
# Pydantic model -> dict
if hasattr(response, "model_dump"):
return response.model_dump()
return response
|