Migrate to aprep_inputs in astream

[edwardzjl]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

In current implementation, AgentExecutorIterator invokes AgentExecutor‘s prep_inputs instead of aprep_inputs:

class AgentExecutorIterator:
    ...
    @inputs.setter
    def inputs(self, inputs: Any) -> None:
        self._inputs = self.agent_executor.prep_inputs(inputs)
    ...

It would be great if we can migrate to aprep_inputs in astream.

Motivation

I'm exploring the PostgresChatMessageHistory and want to utilize asynchronous connections. To do so, I maintained an async posgres connection pool and pass an async connection each time I create an instance of PostgresChatMessageHistory

However, when I wrap the history into a memory and pass that memory to an agent executor, then astream that agent executor, I encounter an error:

Please initialize the PostgresChatMessageHistory with a sync connection or use the async aget_messages method instead.

After some digging, I find that the prep_inputs was called instead of the aprep_inputs, causing PostgresChatMessageHistory to fetch messages synchronously without a sync connection provided.

This can be resolved by passing a sync_connection to PostgresChatMessageHistory during instantiating. However, this requires maintaining a separate synchronize connection pool, which is less than ideal.

Proposal (If applicable)

No response