diff --git a/services/agents/specialists.py b/services/agents/specialists.py index 921124d..be19b2a 100644 --- a/services/agents/specialists.py +++ b/services/agents/specialists.py @@ -1,5 +1,8 @@ from .tools import call_tool from services.index.retrieve import find_precedents +from services.common.breaker import CircuitBreaker + +_pre_breaker = CircuitBreaker(fails=5, reset_after=30) def coverage_node(state: dict) -> dict: @@ -22,8 +25,12 @@ def cost_node(state: dict) -> dict: def precedent_node(state: dict) -> dict: r = state["record"] text = f"{r['incident_type']} {r['severity']} damage. {r['notes_summary']}" - hits = find_precedents(state["image_path"], text, k=5) - return {"precedents": hits} + try: + hits = _pre_breaker.call(find_precedents, state["image_path"], text, k=5) + return {"precedents": hits} + except Exception: + # degraded: no precedents → flag so the synthesiser escalates + return {"precedents": [], "degraded": True} def fraud_node(state: dict) -> dict: diff --git a/services/agents/synthesiser.py b/services/agents/synthesiser.py index 1e15b67..8227a9e 100644 --- a/services/agents/synthesiser.py +++ b/services/agents/synthesiser.py @@ -6,7 +6,7 @@ from .schema import Recommendation load_dotenv() -client = anthropic.Anthropic() +client = anthropic.Anthropic(max_retries=4, timeout=60.0) langfuse = get_client() MODEL = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-5") diff --git a/services/common/breaker.py b/services/common/breaker.py new file mode 100644 index 0000000..ad6e09b --- /dev/null +++ b/services/common/breaker.py @@ -0,0 +1,24 @@ +import time + + +class CircuitBreaker: + """Open after N failures; reject fast; half-open to test recovery.""" + def __init__(self, fails=5, reset_after=30): + self.fails, self.reset_after = fails, reset_after + self.count, self.opened_at = 0, None + + def call(self, fn, *a, **k): + if self.opened_at: # breaker is OPEN + if time.time() - self.opened_at < self.reset_after: + raise RuntimeError("circuit open — dependency unavailable") + self.opened_at = None # half-open: try once + self.count = 0 + try: + r = fn(*a, **k) + self.count = 0 # success resets + return r + except Exception: + self.count += 1 + if self.count >= self.fails: + self.opened_at = time.time() # trip OPEN + raise diff --git a/services/common/resilient.py b/services/common/resilient.py new file mode 100644 index 0000000..6cfa5ca --- /dev/null +++ b/services/common/resilient.py @@ -0,0 +1,13 @@ +from tenacity import (retry, stop_after_attempt, wait_exponential_jitter, + retry_if_exception_type) + +# retry transient network/5xx; backoff 1s,2s,4s... with jitter; 4 tries +external_retry = retry( + stop=stop_after_attempt(4), + wait=wait_exponential_jitter(initial=1, max=10), + retry=retry_if_exception_type((ConnectionError, TimeoutError, Exception)), + reraise=True) + +# usage: decorate the embed + pinecone calls +# @external_retry +# def embed_text(...): ... diff --git a/services/ingest/extractor.py b/services/ingest/extractor.py index 9b2ea8c..6b2ff21 100644 --- a/services/ingest/extractor.py +++ b/services/ingest/extractor.py @@ -19,7 +19,7 @@ load_dotenv() # ── clients & config ─────────────────────────────────────────────── -client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY from env +client = anthropic.Anthropic(max_retries=4, timeout=60.0) # reads ANTHROPIC_API_KEY from env langfuse = get_client() # reads LANGFUSE_* from env MODEL = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-5")