Save `history_processors`'s result for next round model request

[Wh1isper]

Question

Hi, I was migrating my old composer implementation to history_processors based composer and found that I couldn't save the results processed by history_processors, which resulted in each tool return triggering the compact.

I'm now implementing this using a rather tricky way: transfer via ctx with a patched_messages field

	ctx = AgentContext()
    async with agent.iter(..., deps=ctx) as run:
        async for node in run:
            if Agent.is_user_prompt_node(node) or Agent.is_end_node(node):
                continue

            elif Agent.is_model_request_node(node) or Agent.is_call_tools_node(node):
                async with node.stream(run.ctx) as request_stream:
                    # Right after request, save immediately
                    run.ctx.state.message_history = ctx.patched_messages if ctx.patched_messages else ...
                    # Then start streaming
                    async for event in request_stream:
                        ...

Releated

• feat: add history_processors parameter to Agent for message processing #1970
• feat: add RunContext support to history processors #2002

What do you think? I intuit that history_processors will change message history.

cc @Kludex

Additional Context

No response

[DouweM]

@Wh1isper Could you implement your history processor function with a ctx: RunContext argument and handle the ctx.deps.patched_messages read/write there directly?

[Wh1isper]

@Wh1isper Could you implement your history processor function with a ctx: RunContext argument and handle the ctx.deps.patched_messages read/write there directly?

Yes it works well. I'd like to ask if this is a reliable feature or if we might consider making changes to it later?

[Wh1isper]

I created a demo repo: https://github.com/Wh1isper/pydantic-ai-history-processor to explain

Just run main.py https://github.com/Wh1isper/pydantic-ai-history-processor/blob/main/pydantic_ai_history_processor/main.py

[DouweM]

I'd like to ask if this is a reliable feature or if we might consider making changes to it later?

@Wh1isper I think we'd be open to a way to have the history_processor overwrite the cached message history so we don't duplicate effort on every run. But it may be best to wait until @Kludex is back next week.

[AlexEnrique]

I hadn't had time to change my summarization feature to use history_processor yet, but I was expecting that the message_history would be overwritten by the history returned by the last history_processor. So it's not being overwritten, right?

I understand that result.all_messages() will be changed if the history is overwritten, but if we didn't do this, then the history_processor made no difference.

From another perspective, how can I overwrite the message_history if a token threshold is reached in the middle of a run without loosing messages. Because an user might not be using Agent.iter as @Wh1isper did in his demo, it may be using result.new_messages() to save messages in the end of a run, and we can't force the user to move to Agent.iter just because of this.

For now, the only solution I think to solve this problem is to have a callback called after every new ModelRequest or ModelResponse is added to the message_history that could be used to persist the message_history during a run. But I don't know if it is a good idea.

What do you guys think about?

[AlexEnrique]

An idea: to avoid surprises for users using history_processor, could we add a wrapper class HistoryProcessor to be used like this:

agent = Agent(..., history_processors=[HistoryProcessor(my_func, replace_history=True)])

We could also pass single functions, as we do today.

The replace_history param will determine if the message_history should be replaced by the return of the function or not. By default it is false, to keep the current behavior.

---

The question on how to avoid loosing messages when modifying the history during a run is still here.

[Wh1isper]

I prefer HistoryProcessors because it's hard to deal with replace mixups.

Maybe like this:

agent = Agent(..., history_processors=HistoryProcessors([...functions], replace_history=True])

[AlexEnrique]

I prefer HistoryProcessors because it's hard to deal with replace mixups.

Maybe like this:

agent = Agent(..., history_processors=HistoryProcessors([...functions], replace_history=True])

I was thinking about, and there are two problems with this: 1) this may break the current API and 2) this wouldn't conforme with current API, for example Agent(..., tools=[Tool(...)])