History processor replaces message history (#2324)

Co-authored-by: Marcelo Trylesinski <[email protected]>

Author: AlexEnrique

docs/message-history.md

@@ -334,6 +334,10 @@ custom processing logic.
 Pydantic AI provides a `history_processors` parameter on `Agent` that allows you to intercept and modify
 the message history before each model request.
 
+!!! warning "History processors replace the message history"
+    History processors replace the message history in the state with the processed messages, including the new user prompt part.
+    This means that if you want to keep the original message history, you need to make a copy of it.
+
 ### Usage
 
 The `history_processors` is a list of callables that take a list of
@@ -389,6 +393,9 @@ long_conversation_history: list[ModelMessage] = []  # Your long conversation his
 # result = agent.run_sync('What did we discuss?', message_history=long_conversation_history)
 ```
 
+!!! warning "Be careful when slicing the message history"
+    When slicing the message history, you need to make sure that tool calls and returns are paired, otherwise the LLM may return an error. For more details, refer to [this GitHub issue](https://github.com/pydantic/pydantic-ai/issues/2050#issuecomment-3019976269).
+
 #### `RunContext` parameter
 
 History processors can optionally accept a [`RunContext`][pydantic_ai.tools.RunContext] parameter to access
@@ -449,6 +456,9 @@ async def summarize_old_messages(messages: list[ModelMessage]) -> list[ModelMess
 agent = Agent('openai:gpt-4o', history_processors=[summarize_old_messages])
 ```
 
+!!! warning "Be careful when summarizing the message history"
+    When summarizing the message history, you need to make sure that tool calls and returns are paired, otherwise the LLM may return an error. For more details, refer to [this GitHub issue](https://github.com/pydantic/pydantic-ai/issues/2050#issuecomment-3019976269), where you can find examples of summarizing the message history.
+
 ### Testing History Processors
 
 You can test what messages are actually sent to the model provider using

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -365,9 +365,7 @@ async def _prepare_request(
         model_request_parameters = await _prepare_request_parameters(ctx)
         model_request_parameters = ctx.deps.model.customize_request_parameters(model_request_parameters)
 
-        message_history = await _process_message_history(
-            ctx.state.message_history, ctx.deps.history_processors, run_context
-        )
+        message_history = await _process_message_history(ctx.state, ctx.deps.history_processors, run_context)
 
         return model_settings, model_request_parameters, message_history, run_context
 
@@ -859,11 +857,12 @@ def build_agent_graph(
 
 
 async def _process_message_history(
-    messages: list[_messages.ModelMessage],
+    state: GraphAgentState,
     processors: Sequence[HistoryProcessor[DepsT]],
     run_context: RunContext[DepsT],
 ) -> list[_messages.ModelMessage]:
     """Process message history through a sequence of processors."""
+    messages = state.message_history
     for processor in processors:
         takes_ctx = is_takes_ctx(processor)
 
@@ -880,4 +879,7 @@ async def _process_message_history(
             else:
                 sync_processor = cast(_HistoryProcessorSync, processor)
                 messages = await run_in_executor(sync_processor, messages)
+
+    # Replaces the message history in the state with the processed messages
+    state.message_history = messages
     return messages

tests/test_history_processor.py

@@ -5,7 +5,15 @@
 from inline_snapshot import snapshot
 
 from pydantic_ai import Agent
-from pydantic_ai.messages import ModelMessage, ModelRequest, ModelRequestPart, ModelResponse, TextPart, UserPromptPart
+from pydantic_ai.messages import (
+    ModelMessage,
+    ModelRequest,
+    ModelRequestPart,
+    ModelResponse,
+    SystemPromptPart,
+    TextPart,
+    UserPromptPart,
+)
 from pydantic_ai.models.function import AgentInfo, FunctionModel
 from pydantic_ai.tools import RunContext
 from pydantic_ai.usage import Usage
@@ -70,6 +78,71 @@ def no_op_history_processor(messages: list[ModelMessage]) -> list[ModelMessage]:
     )
 
 
+async def test_history_processor_run_replaces_message_history(function_model: FunctionModel):
+    """Test that the history processor replaces the message history in the state."""
+
+    def process_previous_answers(messages: list[ModelMessage]) -> list[ModelMessage]:
+        # Keep the last message (last question) and add a new system prompt
+        return messages[-1:] + [ModelRequest(parts=[SystemPromptPart(content='Processed answer')])]
+
+    agent = Agent(function_model, history_processors=[process_previous_answers])
+
+    message_history = [
+        ModelRequest(parts=[UserPromptPart(content='Question 1')]),
+        ModelResponse(parts=[TextPart(content='Answer 1')]),
+        ModelRequest(parts=[UserPromptPart(content='Question 2')]),
+        ModelResponse(parts=[TextPart(content='Answer 2')]),
+    ]
+
+    result = await agent.run('Question 3', message_history=message_history)
+    assert result.all_messages() == snapshot(
+        [
+            ModelRequest(parts=[UserPromptPart(content='Question 3', timestamp=IsDatetime())]),
+            ModelRequest(parts=[SystemPromptPart(content='Processed answer', timestamp=IsDatetime())]),
+            ModelResponse(
+                parts=[TextPart(content='Provider response')],
+                usage=Usage(requests=1, request_tokens=54, response_tokens=2, total_tokens=56),
+                model_name='function:capture_model_function:capture_model_stream_function',
+                timestamp=IsDatetime(),
+            ),
+        ]
+    )
+
+
+async def test_history_processor_streaming_replaces_message_history(function_model: FunctionModel):
+    """Test that the history processor replaces the message history in the state."""
+
+    def process_previous_answers(messages: list[ModelMessage]) -> list[ModelMessage]:
+        # Keep the last message (last question) and add a new system prompt
+        return messages[-1:] + [ModelRequest(parts=[SystemPromptPart(content='Processed answer')])]
+
+    agent = Agent(function_model, history_processors=[process_previous_answers])
+
+    message_history = [
+        ModelRequest(parts=[UserPromptPart(content='Question 1')]),
+        ModelResponse(parts=[TextPart(content='Answer 1')]),
+        ModelRequest(parts=[UserPromptPart(content='Question 2')]),
+        ModelResponse(parts=[TextPart(content='Answer 2')]),
+    ]
+
+    async with agent.run_stream('Question 3', message_history=message_history) as result:
+        async for _ in result.stream_text():
+            pass
+
+    assert result.all_messages() == snapshot(
+        [
+            ModelRequest(parts=[UserPromptPart(content='Question 3', timestamp=IsDatetime())]),
+            ModelRequest(parts=[SystemPromptPart(content='Processed answer', timestamp=IsDatetime())]),
+            ModelResponse(
+                parts=[TextPart(content='hello')],
+                usage=Usage(request_tokens=50, response_tokens=1, total_tokens=51),
+                model_name='function:capture_model_function:capture_model_stream_function',
+                timestamp=IsDatetime(),
+            ),
+        ]
+    )
+
+
 async def test_history_processor_messages_sent_to_provider(
     function_model: FunctionModel, received_messages: list[ModelMessage]
 ):
@@ -90,7 +163,6 @@ def capture_messages_processor(messages: list[ModelMessage]) -> list[ModelMessag
     assert result.all_messages() == snapshot(
         [
             ModelRequest(parts=[UserPromptPart(content='Previous question', timestamp=IsDatetime())]),
-            ModelResponse(parts=[TextPart(content='Previous answer')], timestamp=IsDatetime()),
             ModelRequest(parts=[UserPromptPart(content='New question', timestamp=IsDatetime())]),
             ModelResponse(
                 parts=[TextPart(content='Provider response')],