kaxil commented on code in PR #74450:
URL: https://github.com/apache/airflow/pull/74450#discussion_r4218073581
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/hitl_review.py:
##########
@@ -181,8 +185,12 @@ def _poll_loop(
key=XCOM_HUMAN_ACTION, task_ids=ti.task_id,
map_indexes=ti.map_index
)
except Exception:
+ consecutive_pull_failures += 1
+ if consecutive_pull_failures >=
_MAX_CONSECUTIVE_XCOM_PULL_FAILURES:
+ raise
Review Comment:
Failures 1 to 9 log a warning, but the 10th re-raises with nothing saying
why, so the task log ends with a raw `xcom_pull` exception that looks like a
one-off error. Since nobody is watching when this fires, a line like
`self.log.error("Giving up HITL review after %d consecutive XCom pull
failures", consecutive_pull_failures)` before the `raise` would tell on-call it
was the HITL loop giving up.
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm.py:
##########
@@ -197,7 +198,7 @@ def __init__(
model_id: str | None = None,
fallback_conn_ids: list[str] | None = None,
system_prompt: str = "",
- output_type: type = str,
+ output_type: OutputSpec[Any] = str,
Review Comment:
`OutputSpec` also admits `[A, B]` and marker forms like `ToolOutput(A)`, and
the description says list forms are accepted now. Those break the approval
resume path though: with `require_approval=True`, `execute_complete` calls
`rehydrate_pydantic_output(self.output_type, ...)`, and `TypeAdapter([A, B])`
raises `PydanticSchemaGenerationError` (a `RuntimeError` subclass) while
`TypeAdapter(ToolOutput(A))` raises `AttributeError`. Neither is in the `except
(ValidationError, ValueError, TypeError)` in `utils/output_type.py`, so the
task fails after the human has approved. `A | B` works fine. Could we either
narrow the annotation to what the operator handles (a type or a union), or
teach `rehydrate_pydantic_output` about sequences and markers? The `:param
output_type:` docstring still says `BaseModel` subclass too.
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/hitl_review.py:
##########
@@ -35,6 +35,9 @@
log = logging.getLogger(__name__)
+# ``hitl_timeout`` may be ``None``, so a persistently failing poll would
otherwise loop forever.
+_MAX_CONSECUTIVE_XCOM_PULL_FAILURES = 10
Review Comment:
The docs still only describe the old exits. The class docstring says the
loop stops after `hitl_timeout` or `max_hitl_iterations`, and
`docs/hitl_review.rst` says `hitl_timeout=None` "blocks until a terminal
action", which stops being true once this limit trips. Could you update both
here, and add the re-raised pull error to the `:raises:` of `run_hitl_review`
while you're in there?
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py:
##########
@@ -160,7 +160,11 @@ def __init__(
ignore_downstream_trigger_rules: bool = False,
**kwargs: Any,
) -> None:
- kwargs.pop("output_type", None)
+ if "output_type" in kwargs:
Review Comment:
`LLMSQLQueryOperator` has the same silent drop (`kwargs.pop("output_type",
None)` in `llm_sql.py`) and `LLMSchemaCompareOperator` overwrites it with
`SchemaCompareResult`. Should those get the same guard? If so, a class-level
flag checked once in `LLMOperator.__init__` might be tidier than repeating the
check in each subclass.
--
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]
