kaxil commented on code in PR #74625:
URL: https://github.com/apache/airflow/pull/74625#discussion_r4266477103
##########
providers/common/ai/docs/message_history.rst:
##########
@@ -47,14 +47,66 @@ straight back in on the next run. When pulling it via a
template, pass
``default='[]'`` (as above) so the first run -- which has no XCom yet --
starts a
fresh session instead of trying to parse the string ``"None"``.
-The transcript is **cumulative**: each turn appends to it, so it grows for the
-life of the session. For long sessions, configure an object-storage XCom
backend
-or trim older turns before the next run rather than feeding the whole history
-back unbounded.
-
.. note::
``message_history`` cannot be combined with ``enable_hitl_review`` -- the
operator raises at construction. The post-review (human-approved)
transcript
is not recoverable today, so emitting the pre-review transcript would
silently drop the reviewed turns.
+
+Message format, and limiting session size
+-----------------------------------------
+
+The transcript is stored as pydantic-ai's JSON message format. It is
+**cumulative**: each turn appends to it, so it grows for the life of the
+session. The operator does not trim it itself, but ``capabilities=`` is passed
+to the pydantic-ai ``Agent``, so the ``ProcessHistory``
+capability from pydantic-ai's documentation works on ``AgentOperator`` as is:
+
+.. code-block:: python
+
+ from pydantic_ai.capabilities import ProcessHistory
+ from pydantic_ai.messages import ModelMessage, ModelRequest, ToolReturnPart
+
+
+ def keep_recent(messages: list[ModelMessage]) -> list[ModelMessage]:
+ tail = messages[-10:]
+ # Start at a request without tool returns, so no tool return is left
+ # without its tool call.
+ for i, message in enumerate(tail):
+ if isinstance(message, ModelRequest) and not any(
+ isinstance(part, ToolReturnPart) for part in message.parts
Review Comment:
In a long tool loop with a tool retry in it, the first match here can be a
request that holds only a tool-bound `RetryPromptPart`. The trimmed history
then starts mid-loop without the user's question, and because the processed
list is written back, the stored transcript loses the question too. pydantic-ai
drops the orphaned retry, so nothing errors and the agent just loses its task
(reproduced with a `FunctionModel` on 2.33.0 and 2.48.0). Checking
`isinstance(part, (ToolReturnPart, RetryPromptPart))` fixes it. The comment
above could then say the goal is to start at a user turn, since pydantic-ai
already removes orphaned tool results.
##########
providers/common/ai/docs/message_history.rst:
##########
@@ -47,14 +47,66 @@ straight back in on the next run. When pulling it via a
template, pass
``default='[]'`` (as above) so the first run -- which has no XCom yet --
starts a
fresh session instead of trying to parse the string ``"None"``.
-The transcript is **cumulative**: each turn appends to it, so it grows for the
-life of the session. For long sessions, configure an object-storage XCom
backend
-or trim older turns before the next run rather than feeding the whole history
-back unbounded.
-
.. note::
``message_history`` cannot be combined with ``enable_hitl_review`` -- the
operator raises at construction. The post-review (human-approved)
transcript
is not recoverable today, so emitting the pre-review transcript would
silently drop the reviewed turns.
+
+Message format, and limiting session size
+-----------------------------------------
+
+The transcript is stored as pydantic-ai's JSON message format. It is
+**cumulative**: each turn appends to it, so it grows for the life of the
+session. The operator does not trim it itself, but ``capabilities=`` is passed
+to the pydantic-ai ``Agent``, so the ``ProcessHistory``
+capability from pydantic-ai's documentation works on ``AgentOperator`` as is:
+
+.. code-block:: python
+
+ from pydantic_ai.capabilities import ProcessHistory
+ from pydantic_ai.messages import ModelMessage, ModelRequest, ToolReturnPart
+
+
+ def keep_recent(messages: list[ModelMessage]) -> list[ModelMessage]:
+ tail = messages[-10:]
+ # Start at a request without tool returns, so no tool return is left
+ # without its tool call.
+ for i, message in enumerate(tail):
+ if isinstance(message, ModelRequest) and not any(
+ isinstance(part, ToolReturnPart) for part in message.parts
+ ):
+ return tail[i:]
+ return messages
+
+
+ AgentOperator(
+ task_id="chat",
+ ...,
+ capabilities=[ProcessHistory(keep_recent)],
+ )
+
+pydantic-ai writes the processed list back into the run state, and
+``result.all_messages()`` returns that list. With a history processor set, the
+transcript pushed to XCom is therefore the **trimmed** one, and turns the
+processor drops are no longer part of the stored session. Alternatively, read
the
+``message_history`` XCom in a downstream task, trim it, store it in your own
+session store, and load it from there on the next run; or configure an
+object-storage XCom backend for long sessions.
+
+The operator has no built-in cross-session memory. To let an agent use facts
Review Comment:
#74681 adds a `memory` page for this case: pydantic-ai-harness `Memory`
passed through `capabilities=`. Whichever of the two merges second could link
it from here, for example "To let the agent keep its own notes across runs, see
:doc:`memory`."
--
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]
