feat(agent): implement history processors for PydanticAI integration …

[Leoyzen]

This PR enables users to configure pydantic-ai history processors in their YAML configuration, allowing dynamic manipulation of message history before LLM requests.

Motivation

Currently, there's no way for users to customize how message history is processed before being sent to the LLM. This feature allows users to define custom processing pipelines (e.g., filtering, summarization, token management) via simple import paths in YAML configuration.

Changes

1. Configuration Schema (src/agentpool_config/session.py)

Added history_processors: list[str] | None field to MemoryConfig:

agents:
  my_agent:
    type: native
    model: openai:gpt-4o-mini
    memory:
      history_processors:
        - my_module:filter_system_messages
        - my_module:truncate_old_messages

• Execution Order: CompactionPipeline runs first, then history processors
• Security: Documented security considerations (processors have full message access)

2. Native Agent Integration (src/agentpool/agents/native_agent/agent.py)

• Dynamic Import Resolution: Processes import paths (e.g., module:function) at agent initialization
• Signature Validation: Validates processor signatures using inspect.signature, supporting all 4 pydantic-ai patterns:
  • (messages) -> messages
  • (ctx, messages) -> messages
  • async (messages) -> messages
  • async (ctx, messages) -> messages
• Caching: Added _resolved_history_processors cache to avoid repeated imports during agentlet creation
• Integration: Resolved processors passed directly to PydanticAgent

3. Comprehensive Test Suite (tests/test_history_processors.py)

15 tests covering:

• Import resolution and caching behavior
• Signature validation and error handling
• Sequential execution of multiple processors
• Interaction with CompactionPipeline and RunContext
• Combined usage with CompactionPipeline (verifies execution order)

Usage Example

# my_processors.py
def filter_thinking_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Remove thinking blocks before sending to LLM."""
    return [m for m in messages if not m.has_thinking_content()]

# agents.yml
agents:
  coder:
    type: native
    model: openai:gpt-4o-mini
    memory:
      history_processors:
        - my_processors:filter_thinking_messages

Execution Flow

User Message
    ↓
CompactionPipeline (runs first - handles compaction/summarization)
    ↓
History Processors (custom user-defined transformations)
    ↓
LLM Request

Testing

All tests pass:

uv run pytest tests/test_history_processors.py -v
# 15 passed

[gemini-code-assist]

Summary of Changes

Hello @Leoyzen, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a powerful new feature enabling users to configure custom history processors for PydanticAI agents. This allows for dynamic and programmable manipulation of message history, such as filtering, summarization, or token management, directly through YAML configuration. The change significantly enhances the flexibility and customization capabilities of the agent's interaction with Large Language Models.

Highlights

• Configurable History Processors: Introduced a new history_processors field in MemoryConfig to allow users to specify custom functions for manipulating message history before LLM requests.
• Dynamic Processor Integration: Implemented dynamic import, signature validation, and caching of user-defined history processors within the NativeAgent to ensure correct and efficient execution.
• Comprehensive Testing: Added a thorough test suite covering various scenarios, including import resolution, signature validation, caching, sequential execution, and interaction with existing history.
• Detailed Documentation: Provided extensive documentation for history processors, including usage examples, execution order relative to CompactionPipeline, and critical security considerations.

Changelog

• src/agentpool/agents/native_agent/agent.py
  • Imported the inspect module for runtime introspection and signature validation.
  • Added _resolved_history_processors as a cache for imported processor callables to prevent redundant imports.
  • Implemented _validate_processor_signature to enforce correct function signatures for history processors (supporting sync/async with/without RunContext).
  • Created _resolve_history_processors to dynamically import processor functions from string paths, validate them, and store them in the cache.
  • Integrated the resolution and passing of history_processors to the PydanticAgent constructor during agentlet creation.
• src/agentpool_config/session.py
  • Added a new optional field history_processors: list[str] | None to MemoryConfig.
  • Included detailed documentation for the history_processors field, outlining its purpose, supported signatures, execution order relative to CompactionPipeline, and important security considerations.
• tests/test_history_processors.py
  • Added a new test file dedicated to validating history processor functionality.
  • Includes tests for MemoryConfig validation with empty or None processor lists.
  • Contains tests for handling invalid import paths, non-callable imports, and incorrect processor signatures.
  • Verifies successful resolution of various valid processor types (sync/async, with/without context).
  • Tests the caching mechanism for history processors to ensure they are resolved only once.
  • Includes integration tests to confirm processors are called during agent execution and run in the correct sequence.
  • Ensures compatibility when no history processors are configured.
  • Validates that history processors receive the complete message history, including pre-existing conversation messages.
• tests/test_processors.py
  • Added a new file containing mock history processor functions used for testing purposes.
  • Includes examples of valid sync and async processors, both with and without RunContext parameters.
  • Provides examples of intentionally invalid processor functions to test signature validation logic.

Activity

• All 15 new tests for history processors are passing, ensuring the stability and correctness of the new feature.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request introduces a valuable feature for configuring pydantic-ai history processors via YAML, enabling dynamic manipulation of message history. The implementation is well-documented, especially the security considerations and usage examples in src/agentpool_config/session.py. The accompanying test suite is comprehensive and covers a wide range of scenarios, ensuring the feature's robustness. I have one suggestion to improve code clarity and reduce redundancy.