diff --git a/README.md b/README.md index c84363f..9b664b4 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ with agentcompile.conversation(ticket.id): ## What happens on each call -1. With a conversation id, the SDK asks AgentCompile for a decision (`/v1/decide`). The whole call takes at most `timeout` (default 2 s), and AgentCompile is told that, so it answers in time. +1. With a conversation id, the SDK asks AgentCompile for a decision (`/v1/decide`). The whole call takes at most `timeout` (default 5 s), and AgentCompile is told that, so it answers in time. 2. **compiled**: the answer (a tool call or a reply) comes back in exactly the provider's response shape; your loop runs the tool as usual. A tool call is only ever one of the `tools` you passed. 3. **forwarded**: your model is called as normal. 4. **fail-open**: AgentCompile errored, was slow or sent something unusable, so your model is called as normal. After 5 failures in a row the SDK stops asking for 30 s, then tries one call. diff --git a/src/agentcompile/__init__.py b/src/agentcompile/__init__.py index 5c47fe4..2495264 100644 --- a/src/agentcompile/__init__.py +++ b/src/agentcompile/__init__.py @@ -65,7 +65,7 @@ def wrap( key: str | None = None, base_url: str | None = None, mode: str = LIVE, - timeout: float = 2.0, + timeout: float = 5.0, trail: str | bool | None = True, on_event: OnEvent | None = None, company: str | None = None, @@ -145,7 +145,7 @@ def transport( base_url: str | None = None, company: str | None = None, scrub: bool = True, - timeout: float = 2.0, + timeout: float = 5.0, ) -> Any: """A transport that captures every model call through it, for frameworks whose client you don't hold (LangChain, LiteLLM, CrewAI, Pydantic AI, ...). Capture only: calls are never @@ -165,7 +165,7 @@ def async_transport( base_url: str | None = None, company: str | None = None, scrub: bool = True, - timeout: float = 2.0, + timeout: float = 5.0, ) -> Any: """The async version of transport().""" _, async_ = classes(lib_of(wrapped, lib)) diff --git a/src/agentcompile/_capture.py b/src/agentcompile/_capture.py index b930991..12609f0 100644 --- a/src/agentcompile/_capture.py +++ b/src/agentcompile/_capture.py @@ -2,8 +2,10 @@ can find the jobs your agent repeats (opt-in: `wrap(..., capture=True)`). Never in the way: calls are queued and sent by one background thread in batches; a full queue -drops the oldest, a failed send is dropped and counted, and nothing here ever raises into your -agent. Each record is one line in the exchange-log shape AgentCompile's importer reads: +drops the oldest, a send that fails for a passing reason (no connection, a timeout, 408, 429, +5xx) is retried a few times with backoff, one that still fails is dropped and counted, and +nothing here ever raises into your agent. Each record is one line in the exchange-log shape +AgentCompile's importer reads: {"provider", "conversation_id", "timestamp", "request", "response"}. """ @@ -31,6 +33,12 @@ MAX_BATCH_BYTES = 4 * 1024 * 1024 MAX_RECORD_BYTES = 7 * 1024 * 1024 INTERVAL = 1.0 +# A failed batch is tried RETRIES more times, waiting BACKOFF, then twice that, ... (or what +# the server's Retry-After asks, up to MAX_WAIT); about 3.5 s in all before it is dropped. +RETRIES = 3 +BACKOFF = 0.5 +MAX_WAIT = 10.0 +_PASSING = frozenset({408, 425, 429, 500, 502, 503, 504}) _ALL: dict[tuple[Any, ...], Capturer] = {} @@ -72,7 +80,10 @@ def __init__( self._thread: threading.Thread | None = None self.sent = 0 self.dropped = 0 - self.failed = 0 + self.failed = ( + 0 # calls dropped after their send failed (and was retried, if it could be) + ) + self.retried = 0 # batch sends tried again atexit.register(self.flush, 2.0) def add( @@ -147,7 +158,7 @@ def flush(self, timeout: float = 5.0) -> None: """Send everything queued now (tests, and at exit).""" deadline = time.monotonic() + timeout while self._queue and time.monotonic() < deadline: - self._send_batch() + self._send_batch(deadline) def _start(self) -> None: if self._thread is None or not self._thread.is_alive(): @@ -195,7 +206,9 @@ def _put_back(self, rest: list[dict[str, Any]]) -> None: self.dropped += len(rest) - len(keep) self._queue.extendleft(reversed(keep)) - def _send_batch(self) -> None: + def _send_batch(self, deadline: float | None = None) -> None: + """Send one batch, retrying a passing failure with backoff; never past `deadline` (a + flush's), when there is one.""" batch = self._take() if not batch: return @@ -203,14 +216,35 @@ def _send_batch(self) -> None: headers.pop(CONVERSATION_HEADER, None) headers["content-type"] = "application/json" url = url.rsplit("/v1/decide", 1)[0] + "/v1/capture" + body = ('{"exchanges": [' + ", ".join(batch) + "]}").encode("utf-8") + for attempt in range(RETRIES + 1): + wait = self._post(url, headers, body) + if wait is None: + self.sent += len(batch) + return + if wait < 0 or attempt == RETRIES: + break + wait = min(max(wait, BACKOFF * 2**attempt), MAX_WAIT) + if deadline is not None and time.monotonic() + wait > deadline: + break + self.retried += 1 + time.sleep(wait) + self.failed += len(batch) + + def _post(self, url: str, headers: dict[str, str], body: bytes) -> float | None: + """None when the batch was taken; else how long the server asks us to wait before + trying again (0: no preference), or -1 for a failure retrying won't fix.""" try: if self._http is None: self._http = httpx.Client(timeout=self.settings.timeout) - body = '{"exchanges": [' + ", ".join(batch) + "]}" - response = self._http.post(url, headers=headers, content=body.encode("utf-8")) - if response.status_code == 200: - self.sent += len(batch) - else: - self.failed += len(batch) + response = self._http.post(url, headers=headers, content=body) except Exception: - self.failed += len(batch) + return 0.0 + if response.status_code == 200: + return None + if response.status_code not in _PASSING: + return -1.0 + try: + return float(response.headers.get("retry-after", 0)) + except ValueError: + return 0.0 diff --git a/tests/test_capture.py b/tests/test_capture.py index a5b3fcd..ebd3b61 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -120,6 +120,72 @@ def test_a_failing_service_never_breaks_the_call() -> None: ) +def _capturer_against(handler: Any) -> Any: + settings = agentcompile.Settings("http://ac.test", "ack_acme.k", None, 1.0) + capturer = _capture.Capturer(settings, httpx.Client(transport=httpx.MockTransport(handler))) + capturer._start = lambda: None # type: ignore[method-assign] + capturer.add("openai", "c1", {"model": "gpt-x", "messages": MESSAGES}, COMPLETION) + return capturer + + +@pytest.fixture +def _no_waiting(monkeypatch: pytest.MonkeyPatch) -> list[float]: + waits: list[float] = [] + monkeypatch.setattr(_capture.time, "sleep", waits.append) + return waits + + +def test_a_passing_failure_is_retried_until_it_lands(_no_waiting: list[float]) -> None: + answers = iter([503, 429, 200]) + attempts: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if not attempts: + attempts.append(0) + raise httpx.ConnectError("down") + attempts.append(1) + return httpx.Response(next(answers), json={}) + + capturer = _capturer_against(handler) + capturer._send_batch() + assert (capturer.sent, capturer.failed, capturer.retried) == (1, 0, 3) + assert _no_waiting == [0.5, 1.0, 2.0] # backoff doubles + + +def test_retries_are_bounded_and_the_drop_counted(_no_waiting: list[float]) -> None: + calls: list[int] = [] + capturer = _capturer_against(lambda r: calls.append(1) or httpx.Response(502)) + capturer._send_batch() + assert len(calls) == _capture.RETRIES + 1 + assert (capturer.sent, capturer.failed) == (0, 1) + + +def test_a_failure_retrying_wont_fix_is_not_retried(_no_waiting: list[float]) -> None: + calls: list[int] = [] + capturer = _capturer_against(lambda r: calls.append(1) or httpx.Response(401)) + capturer._send_batch() + assert len(calls) == 1 and capturer.failed == 1 and _no_waiting == [] + + +def test_retry_after_is_honoured_up_to_a_cap(_no_waiting: list[float]) -> None: + answers = iter( + [ + httpx.Response(429, headers={"retry-after": "3"}), + httpx.Response(429, headers={"retry-after": "3600"}), + httpx.Response(200), + ] + ) + capturer = _capturer_against(lambda r: next(answers)) + capturer._send_batch() + assert _no_waiting == [3.0, _capture.MAX_WAIT] and capturer.sent == 1 + + +def test_a_flush_never_waits_past_its_deadline(_no_waiting: list[float]) -> None: + capturer = _capturer_against(lambda r: httpx.Response(503)) + capturer.flush(0.1) # the first backoff (0.5 s) would overrun it + assert _no_waiting == [] and capturer.failed == 1 + + def test_a_full_queue_drops_the_oldest(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(_capture, "MAX_QUEUE", 2) settings = agentcompile.Settings("http://ac.test", "ack_acme.k", None, 1.0)