aap_core

Agent design pattern core framework for building, orchestrating, and training AI agents.

What is this?

aap_core is the foundational framework within the AI Agent Pattern project. It provides a unified abstraction layer for building AI agents, orchestrating multi-agent workflows, and training policies via reinforcement learning — all built on Pydantic and compatible with the A2A protocol v1.0 (a2a-sdk>=1.0.0).

The framework separates orchestration logic from LLM manipulation, delegating model interaction to integration packages (aap_langchain, aap_llamaindex, aap_dspy, aap_transformers). This lets you swap backends without changing agent architecture.

Quick Start

pip install aap_core
# or
uv add aap_core

from a2a.types import AgentCard
from aap_core.agent import BaseAgent
from aap_core.types import AgentMessage

# Define a minimal agent
class MyAgent(BaseAgent):
    card: AgentCard = AgentCard(
        name="my-agent",
        description="A simple agent",
        skills=[],
    )

    def execute(self, message: AgentMessage, **kwargs) -> AgentMessage:
        # Your agent logic here
        message.responses = [("my-agent", "Hello!")]
        message.execution_result = "success"
        return message

Features

• Agent Abstraction: Clean BaseAgent interface with A2A AgentCard integration for self-describing agent metadata
• Orchestration Patterns: Built-in agents for reflection, looping, voting, coordination, and debate patterns
• Chain System: Generic multi-turn causal chains with history management, tool processing, and token tracking
• Prompt Augmentation: Data and structure-based prompt enhancement with configurable looping strategies
• Retrieval System: Pluggable retrievers with post-processing (e.g., rerankers, summarizers)
• RL Policy Training: Full RL loop with replay buffers, exploration modules, and trainers (PPO, DPO, GRPO, Reinforce++), powered by Gymnasium
• Thought Chains: Chain-of-Thought, Tree-of-Thought, Graph-of-Thought, and Self-Consistency patterns
• Guardrails: Extensible guardrail system for input/output filtering
• Async Support: All agent execution methods have async counterparts (aexecute)
• Token Usage Tracking: Automatic token accounting across multi-agent chains

Installation

Core

pip install aap_core

Integration Packages

pip install aap_langchain   # LangChain integration
pip install aap_llamaindex  # LlamaIndex integration
pip install aap_dspy        # DSPy integration
pip install aap_transformers  # Hugging Face Transformers integration

Usage

Building an Agent

from aap_core.agent import BaseAgent
from aap_core.types import AgentMessage
from a2a.types import AgentCard

class TaskAgent(BaseAgent):
    def execute(self, message: AgentMessage, **kwargs) -> AgentMessage:
        self.state = "running"
        # Process the message...
        message.responses = [(self.card.name, "Task completed")]
        message.execution_result = "success"
        self.state = "idle"
        return message

Using Orchestration Patterns

from aap_core.orchestration import ReflectionAgent, LoopAgent

# Self-reflection: execute then reflect on the result
reflector = ReflectionAgent(
    chain_task=my_chain,
    chain_reflection=reflection_chain,
)
result = reflector.execute(message)

# Loop until a condition is met
stopper = lambda msg: "done" in msg.query.lower()
loop = LoopAgent(agent=my_agent, stop_condition=stopper)
result = loop.execute(message)

Prompt Augmentation

from aap_core.prompt_augmenter import SimplePromptAugmenter

# Augment prompts with external context data
augmenter = SimplePromptAugmenter(
    format="Context:\n{context}\n\nQuery: {query}",
    loop=3,  # Apply 3 times
)
augmented = augmenter(message)

RL-Based Prompt Optimization

from aap_core.policy_trainer import PPOTrainer, SimpleReplayBuffer
from aap_core.prompt_augmenter import PromptOptimizationEnv

# Create the environment with multiple augmenters
env = PromptOptimizationEnv(augmenters=[augmenter_a, augmenter_b])

# Create a replay buffer and trainer
buffer = SimpleReplayBuffer(capacity=10000)
trainer = PPOTrainer(policy=my_policy, buffer=buffer)
trainer.train(env, num_steps=10000)

Architecture

aap_core is organized into the following modules:

Documentation

• Full project documentation: aap_core docs
• Example notebooks: example
• A2A Protocol: a2a-protocol.org · Python SDK a2aproject/a2a-python

Contributing

See the main project README for contribution guidelines.

License

This project is licensed under the MIT License — see the LICENSE file for details.