kaxil opened a new pull request, #74770:
URL: https://github.com/apache/airflow/pull/74770
When a task running an OpenAI Agents SDK agent fails and Airflow retries it,
the agent starts again from the prompt: every model call is paid for again and
every tool call happens again. `run_durably()` / `run_durably_sync()` take the
same arguments as `Runner.run` / `Runner.run_sync` and record each model call
and function tool call in the task's durable journal, the one
`AgentOperator(durable=True)` uses. A retry replays every call that asks what
the failed try asked, in order, and runs live from the first that differs. Once
the run succeeds the recording is deleted.
```python
@task(retries=3)
def research(question: str) -> str:
agent = Agent(name="analyst", model=model, tools=AirflowTools(warehouse))
return run_durably_sync(agent, question).final_output
```
From Airflow 3.3 the journal is kept in the task state store; before that,
under `[common.ai] durable_cache_path`, as for `AgentOperator(durable=True)`.
Stacked on the tracing PR.
## Design rationale
**Model calls are intercepted by wrapping the model, not by run hooks.**
`RunHooks` and `AgentHooks` return `None`, so they can observe a call but not
replace it, and `RunState.to_json()` captures a run only at an interruption,
not mid-run. The SDK resolves a model from `RunConfig.model`, then the
`model_provider` for names, then a `Model` on the agent, so `run_durably` wraps
the provider and the `run_config` model and clones the agent graph (following
`handoffs`) to wrap `Model` instances and function tools. This is the same
interception point the SDK's Temporal integration uses.
**A replayed call has to turn into the same next request as the live one
did.** Each request is fingerprinted from the model, instructions,
conversation, settings, tools (with descriptions and strictness), output
schema, handoffs and prompt, and four things keep that stable across a replay:
- Recorded answers keep only the fields the response set (`exclude_unset`),
and `None` fields are left out of the fingerprint. Otherwise replayed items
carry explicit `null`s the live ones do not, and every request after the first
replay mismatches.
- Tool results are recorded as the SDK sends them to the model: structured
`ToolOutput*` content as it is, anything else as `str(output)`. Recording a
Pydantic model, tuple or datetime as JSON would replay as different text.
- The `prompt_cache_key` the SDK generates once per run is left out of the
settings.
- A call the SDK's own model retry makes again reuses the failed call's
position instead of taking a new one, so the next try still lines up.
**An agent used through `as_tool` is one recorded call.** Its own model
calls run inside the outer tool call, so they run live rather than taking
positions of the outer run; the outer call replays whole.
**A run the model failed asks the model again.** For `MaxTurnsExceeded`,
`ModelBehaviorError`, `ModelRefusalError` and output or tool-input guardrail
tripwires the recorded answers would replay into the same failure, so the
journal reruns from the model's last request.
**Server-side conversations are refused.** `previous_response_id`,
`conversation_id` and `auto_previous_response_id` keep the conversation on
OpenAI's servers, which a replayed call never reaches, so `run_durably` raises
`ValueError` for them.
## Known issues
- Not recorded, so they run live on every try: tools other than function
tools (hosted tools, `mcp_servers` tools, shell and computer tools), a function
tool whose `needs_approval` is a function (the SDK checks the approved
arguments only through its own invoker, so wrapping it would weaken that
check), an agent reached through `handoff()` unless its model is given by name,
and streamed runs. A `session` makes every retry run live. The guide lists all
of these.
- A `ModelBehaviorError` that a tool raises, such as arguments failing
validation with `failure_error_function=None`, counts as the tool's failure:
the tool reruns, but the answer that asked for it replays.
- Small follow-ups: passing `result.last_agent` back into `run_durably`
wraps it twice; the wrapped model provider is not closed when `run_config` is
omitted, which matters only for the 0.23 websocket transport; a dict
`run_config` is not accepted.
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
