chore(detectors): added async detectors execution (WIP) - #277
Conversation
github-actions
Bot
added
area-detectors
poshinchen
changed the title
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| *, | ||
| model: Model | str | None = None, | ||
| confidence_threshold: ConfidenceLevel = ConfidenceLevel.LOW, | ||
| max_workers: int = 5, |
There was a problem hiding this comment.
Issue: max_workers carries two different meanings across the public async API. In detect_failures_async, analyze_root_cause_async, and diagnose_session_async, max_workers caps inner per-session LLM concurrency. Here in diagnose_sessions_async, max_workers caps session-level concurrency, and the inner cap is a separate per_session_workers arg. A caller who learned the meaning from diagnose_session_async will misread this signature.
Suggestion: Rename the batch-level knob to something unambiguous (e.g. max_concurrent_sessions) so the two axes of concurrency read clearly, keeping per_session_workers for the inner cap. This also makes the obvious path the happy path (CONTRIBUTING tenet #4).
| def _detect_direct(user_prompt: str, model: Model, template: object) -> list[FailureItem]: | ||
| """Attempt direct LLM detection on the full session.""" | ||
| text = _call_model(model, system_prompt=template.SYSTEM_PROMPT, user_prompt=user_prompt) | ||
| async def _detect_direct_async(user_prompt: str, model: Model, template: object) -> list[FailureItem]: |
There was a problem hiding this comment.
Issue: template: object discards type information — _detect_direct_async / _detect_chunked_async then access template.SYSTEM_PROMPT and template.build_prompt(...), which only type-checks because attr-defined is globally disabled in the mypy config. This loses the safety net for these calls.
Suggestion: Annotate with the concrete template type returned by get_template(...) (and add a return annotation to get_template if it lacks one). That restores attribute checking and documents the contract for readers.
| *, | ||
| model: Model | str | None = None, | ||
| confidence_threshold: ConfidenceLevel = ConfidenceLevel.LOW, | ||
| max_workers: int = 5, |
There was a problem hiding this comment.
Issue: The default max_workers=5 (and per_session_workers=5) is hard-coded as a literal in four separate functions across diagnosis.py, failure_detector.py, and root_cause_analyzer.py. If the sensible default ever needs tuning, it has to be changed in several places and can drift out of sync.
Suggestion: Define a single shared constant (e.g. DEFAULT_MAX_WORKERS in detectors/constants.py) and reference it from each signature.
| for t in pending: | ||
| t.cancel() | ||
| await asyncio.gather(*pending, return_exceptions=True) | ||
| for t in tasks: |
There was a problem hiding this comment.
Issue: The docstring says the first observed exception surfaces, but asyncio.wait(FIRST_EXCEPTION) can return with several tasks already done, and the subsequent loop re-raises the first task in list order that has an exception — not necessarily the one that completed first. The test test_bounded_gather_fail_fast_first_exception_wins only passes because the list-first task also fails first (delay 0.0), so it doesn't actually distinguish the two semantics.
Suggestion: Either tighten the wording to "the first failing task in submission order" or, if temporal ordering matters, capture the exception from the task that triggered the FIRST_EXCEPTION return rather than scanning in list order.
| `return_exceptions=True`, failed tasks appear as `BaseException` | ||
| instances rather than raising. | ||
| """ | ||
| if max_workers < 1: |
There was a problem hiding this comment.
Issue: bounded_gather accepts an Iterable[Awaitable] and, when max_workers < 1, raises after the caller has already created the coroutines (the coros generator/list is materialized by the caller). Any already-constructed coroutines that are never awaited will emit "coroutine was never awaited" warnings. The same applies to bounded_gather_fail_fast. The tests only exercise the invalid case with an empty iterable, so this leak path is untested.
Suggestion: Either validate max_workers before constructing coroutines is the caller's responsibility (document it), or close/drain the passed-in awaitables before raising. At minimum, add a test that passes non-empty coros with max_workers=0 to lock in the intended behavior.
|
|
||
| await diagnose_sessions_async([_make_session("x")], per_session_workers=7) | ||
|
|
||
| assert mock_detect.call_args.kwargs["max_workers"] == 7 |
There was a problem hiding this comment.
Issue: This suite covers per_session_workers threading nicely, but there's no test asserting the session-level concurrency cap (max_workers) actually bounds how many sessions run at once. Given that this PR's whole purpose is bounded concurrency, that cap is the core contract for this function and is currently unverified.
Suggestion: Add a test (mirroring test_detect_failures_chunked_concurrent_calls) that tracks peak in-flight sessions and asserts it never exceeds max_workers.
|
Issue (process / API review): This PR adds several new public APIs ( Suggestion: Before marking ready-for-review, fill in the description with: motivation, the new public signatures (with defaults), example usage for the common batch case ( |
|
Assessment: Comment (PR is marked WIP) Solid, well-tested async refactor — the sync→async split is clean, Review Categories
Nice work on the concurrency test coverage — the peak-in-flight and cancellation assertions are exactly the right things to lock in. |
1 participant
Description
Adds async support to the
detectors/module so failure detection, root cause analysis, and session diagnosis can run concurrently and integrate cleanly with async callers (Experiment._run_diagnosis, FastAPI handlers, Jupyter, async tests).New async API surface (re-exported from
strands_evals.detectors):detect_failures_async(..., max_workers=5)—failure_detector.pyanalyze_root_cause_async(..., max_workers=5)—root_cause_analyzer.pydiagnose_session_async(..., max_workers=5)—diagnosis.pydiagnose_sessions_async(sessions, ..., max_workers=5, per_session_workers=5)— batch helper for many sessionsSync APIs preserved.
detect_failures/analyze_root_cause/diagnose_sessionremain available with their original signatures and now route throughrun_async(lambda: …_async(…)). That means they're safe to call from sync code and from inside a running event loop (Jupyter, FastAPI, async tests) — the work runs on a dedicated worker thread with its own loop, so noRuntimeError: asyncio.run() cannot be called from a running event loop.Internal fan-out.
_detect_chunked_asyncand_rca_chunked_asyncrun chunks/windows concurrently under amax_workerssemaphore via two new helpers in_async.py:bounded_gather(coros, max_workers, return_exceptions=True)— sibling isolation; used bydiagnose_sessions_asyncso one bad session yieldsNoneinstead of poisoning the batch.bounded_gather_fail_fast(coros, max_workers)— usesasyncio.wait(..., FIRST_EXCEPTION), cancels still-running siblings, drains, and re-raises. Used by chunked detection / RCA. Both already swallowContextWindowOverflowException/_is_context_exceededas a successful empty result, so only genuine errors trigger the cancel — which is exactly when we want to stop spending tokens on doomed sibling work.Removed dead bridges.
failure_detector._call_modelis gone; the oldrun_asyncthread bridge is no longer needed there becausemodel.stream(...)is awaited natively. RCA's three syncAgent(...)(prompt, structured_output_model=...)call sites now useawait agent.invoke_async(...).Experiment integration.
Experiment._run_diagnosis(experiment.py:474) swappedawait asyncio.to_thread(diagnose_session, ...)→await diagnose_session_async(...)— drops the redundant thread hop.DiagnosisConfiggainsmax_workers: int = Field(default=5, ge=1). The value is forwarded intodiagnose_session_asyncso users can tune per-session concurrency without monkey-patching.Related Issues
n/a
Documentation PR
n/a
Type of Change
New feature (additive — sync API and existing behavior unchanged).
Testing
hatch run prepare— passes (ruff format, ruff lint, mypy, full pytest).New tests:
tests/strands_evals/test_async.py—bounded_gatherorder/isolation +bounded_gather_fail_fastorder, sibling cancellation on first error, first-exception-wins semantics, empty/invalid inputs.tests/strands_evals/detectors/test_failure_detector.py— async happy path,max_workersforwarded, chunk fan-out actually overlaps (peak ≥ 2 in flight), one-overflow-doesn't-poison-batch, non-context error propagates, fail-fast cancels slow sibling, sync wrapper works inside a running event loop.tests/strands_evals/detectors/test_root_cause_analyzer.py— async direct path, empty failures shortcut, auto-detection threadsmax_workers, sync wrapper works inside a running event loop.tests/strands_evals/detectors/test_diagnosis.py—diagnose_session_asynchappy path,max_workersthread-through;diagnose_sessions_asyncbatch order, one-bad-session→None,per_session_workersthread-through; sync wrapper works inside a running event loop.tests/strands_evals/test_experiment.py—DiagnosisConfig.max_workersreachesdiagnose_session_async. The pre-existingtest_diagnosis_exception_returns_nonemock target updated to the async name.Existing detector tests that mocked the old sync hooks were updated to use
AsyncMockagainst_call_model_async/Agent.invoke_async/detect_failures_async.hatch run prepareChecklist
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.