-
Notifications
You must be signed in to change notification settings - Fork 615
fix: drop redundant on_trace kwarg from Agent.run #2091
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -222,18 +222,16 @@ async def run( | |
| *, | ||
| runtime: Runtime | None = None, | ||
| tools: Mapping[str, SharedToolServer] | None = None, | ||
| on_trace: Callable[[Trace], None] | None = None, | ||
| ) -> Trace: | ||
| """Run this agent on `task` once and return the trace: `runtime` places it | ||
| into a live borrowed box instead of provisioning one; `tools` are live | ||
| servers borrowed from their owner, counted in the pairing check; | ||
| `on_trace` observes the trace the moment it's minted, before any I/O. | ||
| servers borrowed from their owner, counted in the pairing check. | ||
| Retries whole while the trace ends with a retryable error (`config.retries`) | ||
| — never into a borrowed box; the final trace keeps earlier attempts' errors.""" | ||
| retry = self.config.retries | ||
| history: list = [] | ||
| for attempt in range(retry.max_retries + 1): | ||
| trace = await self._run_once(task, runtime, tools, on_trace) | ||
| trace = await self._run_once(task, runtime, tools) | ||
| if attempt == retry.max_retries or not trace_should_retry(trace, retry): | ||
| break | ||
| if runtime is not None: | ||
|
|
@@ -243,6 +241,8 @@ async def run( | |
| ) | ||
| break | ||
| history.extend(trace.errors) | ||
| # This attempt is abandoned; drop it from any live view before backoff. | ||
| self._discard(trace) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Live view pending during retryMedium Severity Calling Reviewed by Cursor Bugbot for commit 408b45a. Configure here. |
||
| delay = backoff(attempt) | ||
| logger.warning( | ||
| "retrying agent rollout (retry %d/%d) in %.1fs after error: %s", | ||
|
|
@@ -258,15 +258,22 @@ async def run( | |
| trace.errors = history + trace.errors | ||
| return trace | ||
|
|
||
| def _observe(self, trace: Trace) -> None: | ||
| """The moment a trace is minted, before any I/O. Base: no-op; an episode | ||
| seat stamps its standing onto the trace and feeds the episode's live view.""" | ||
|
|
||
| def _discard(self, trace: Trace) -> None: | ||
| """A retry is abandoning this attempt's trace. Base: no-op; an episode seat | ||
| drops it from live views (only the final attempt joins the episode).""" | ||
|
|
||
| async def _run_once( | ||
| self, | ||
| task: Task, | ||
| runtime: Runtime | None, | ||
| shared_tools: Mapping[str, SharedToolServer] | None, | ||
| on_trace: Callable[[Trace], None] | None, | ||
| ) -> Trace: | ||
| params = self._rollout_params(task, runtime, dict(shared_tools or {})) | ||
| run = RolloutRun(task=task, on_trace=on_trace, **params) | ||
| run = RolloutRun(task=task, on_trace=self._observe, **params) | ||
| try: | ||
| if await run.open(): | ||
| await run.step() | ||
|
|
@@ -387,37 +394,33 @@ def __init__( | |
| def _shared_for(self, task: Task) -> Mapping[str, SharedToolServer]: | ||
| return self._shared_tools if isinstance(task, self._task_cls) else {} | ||
|
|
||
| def _observe(self, trace: Trace) -> None: | ||
| # The seat's standing lands on the trace the moment it's minted, then the | ||
| # trace joins the episode's live view (before any I/O, retries included). | ||
| if trace.agent is not None: | ||
| trace.agent.name = self._name | ||
| trace.agent.trainable = self.trainable | ||
| if self._on_trace is not None: | ||
| self._on_trace(trace) | ||
|
|
||
| def _discard(self, trace: Trace) -> None: | ||
| # A per-agent retry abandons this attempt's trace; drop it from live views | ||
| # (only the final attempt joins the episode). | ||
| if self._on_discard is not None: | ||
| self._on_discard(trace) | ||
|
|
||
| async def run( | ||
| self, | ||
| task: Task, | ||
| *, | ||
| runtime: Runtime | None = None, | ||
| tools: Mapping[str, SharedToolServer] | None = None, | ||
| on_trace: Callable[[Trace], None] | None = None, | ||
| ) -> Trace: | ||
| last: Trace | None = None | ||
|
|
||
| def watch(trace: Trace) -> None: | ||
| nonlocal last | ||
| if trace.agent is not None: | ||
| trace.agent.name = self._name | ||
| trace.agent.trainable = self.trainable | ||
| # A per-agent retry mints a replacement: the abandoned attempt's trace | ||
| # must leave live views (only the final one joins the episode). | ||
| if last is not None and self._on_discard is not None: | ||
| self._on_discard(last) | ||
| last = trace | ||
| if self._on_trace is not None: | ||
| self._on_trace(trace) | ||
| if on_trace is not None: | ||
| on_trace(trace) | ||
|
|
||
| async with self._gate or nullcontext(): | ||
| trace = await super().run( | ||
| task, | ||
| runtime=runtime, | ||
| tools=tools if tools is not None else self._shared_for(task), | ||
| on_trace=watch, | ||
| ) | ||
| self._completed.append(trace) | ||
| return trace | ||
|
|
||


Uh oh!
There was an error while loading. Please reload this page.