Custom Agents
Build specialized agents with custom logic and behavior
Custom agents provide ultimate flexibility by allowing you to define arbitrary orchestration logic. They go beyond predefined workflow patterns to enable highly specific and complex agent behaviors.
Advanced Concept
Building custom agents requires implementing core execution logic directly. We recommend understanding LLM agents and workflow agents first before tackling custom orchestration.
What Are Custom Agents?
Custom agents inherit from BaseAgent
and implement their own control flow by defining the _run_async_impl
method. You have complete control over:
- How sub-agents are called and orchestrated
- State management and data flow
- Conditional logic and decision making
- External integrations and API calls
When to Use Custom Agents
Consider custom agents when you need:
- Conditional Logic: Different execution paths based on runtime conditions
- Complex State Management: Intricate logic for maintaining workflow state
- External Integrations: Direct API calls, database operations, or custom libraries
- Dynamic Agent Selection: Choosing sub-agents based on runtime evaluation
- Unique Patterns: Orchestration logic that doesn't fit standard patterns
Design Decision
Use custom agents when predefined workflow agents (Sequential, Parallel, Loop) cannot express your required logic patterns.
Core Implementation
The _run_async_impl
Method
This is the heart of every custom agent:
- Asynchronous Generator: Must yield events back to the runner
- Context Access: Receive
InvocationContext
for session and state access - Event Handling: Process and forward events from sub-agents
- Custom Logic: Implement any required control flow
Key Capabilities
🔄 Sub-Agent Orchestration
Call and coordinate multiple sub-agents with custom logic
📊 State Management
Read from and write to session state for data sharing
🧠 Control Flow
Use Python constructs for conditional and iterative workflows
🔌 External Integration
Integrate APIs, databases, and custom services directly
Design Patterns
Conditional Workflows
Execute different agent paths based on runtime conditions or previous results.
State-Driven Orchestration
Use session state to coordinate complex data flows between multiple agents.
External Service Integration
Combine agent capabilities with external APIs and data sources.
Dynamic Agent Selection
Choose which agents to execute based on input analysis or environment conditions.
Managing Sub-Agents
Agent Hierarchy
- Store sub-agents as instance attributes during initialization
- Declare sub-agents in the
sub_agents
list for framework awareness - Use direct method calls for orchestration within your custom logic
State Coordination
- Use
ctx.session.state
for data sharing between agents - Leverage
output_key
parameters for structured data flow - Implement state validation and error handling
Event Processing
- Yield events from sub-agents to maintain streaming behavior
- Optionally inspect, filter, or transform events
- Handle error events and implement recovery logic
Best Practices
Design Considerations
- Keep custom logic focused and well-documented
- Design clear interfaces between sub-agents
- Implement proper error handling and recovery
- Test complex workflows thoroughly with various scenarios
Architecture Design
- Break complex workflows into logical sub-agents
- Use clear naming conventions for state keys
- Document the expected data flow and dependencies
- Design for testability and maintainability
Error Handling
- Implement graceful degradation for agent failures
- Use try-catch blocks around critical operations
- Provide meaningful error messages and logging
- Consider retry logic for transient failures
Performance Considerations
- Minimize unnecessary state operations
- Consider async/await patterns for external calls
- Implement timeouts for long-running operations
- Monitor resource usage in complex workflows
Testing Custom Agents
Unit Testing
- Test individual logic components separately
- Mock sub-agents for isolated testing
- Verify state management logic
- Test error conditions and edge cases
Integration Testing
- Test complete workflows end-to-end
- Verify state flow between agents
- Test with realistic data scenarios
- Validate error handling and recovery
Performance Testing
- Measure execution times for complex workflows
- Test resource usage under load
- Verify timeout and limit behaviors
- Monitor memory usage patterns
Common Use Cases
Multi-Stage Content Generation
Create content through generation, critique, revision, and validation stages with conditional regeneration.
Data Processing Pipelines
Implement complex ETL workflows with conditional transformations and quality checks.
Decision Support Systems
Build agents that analyze data, consult multiple sources, and provide structured recommendations.
Quality Assurance Workflows
Implement multi-step validation processes with conditional approval and revision cycles.