Built-in Agents

AgentMap provides a comprehensive set of built-in agents for common workflow tasks. All agents inherit from BaseAgent and implement a process() method for data transformation.

Agent Architecture

BaseAgent Foundation

Every AgentMap agent follows the same architectural pattern:

class BaseAgent:
    def __init__(self, name: str, prompt: str, context: Dict[str, Any] = None):
        self.name = name
        self.prompt = prompt
        self.context = context or {}
        self.input_fields = self.context.get("input_fields", [])
        self.output_field = self.context.get("output_field", None)
    
    def process(self, inputs: Dict[str, Any]) -> Any:
        """Transform inputs to output - implemented by each agent type"""
        raise NotImplementedError("Subclasses must implement process()")

Key Principles:

• Single Responsibility: Each agent has one clear purpose
• Data Flow: Input fields → process() method → output field
• Error Handling: Built-in exception handling and error routing
• Service Injection: Optional services (LLM, storage) injected via protocols

Agent Execution Flow

1. Input Extraction: System extracts values from state using input_fields
2. Processing: Agent's process() method transforms inputs
3. Output Integration: Result stored in state using output_field
4. Flow Control: Success routes to next_node, errors to error_node

Agent Categories

🧠 LLM Agents

AI-powered agents using language models for reasoning and generation:

Agent Type	Purpose	Key Features
llm	General LLM processing	Multi-provider support, routing, memory
summary	Text summarization	Configurable length, format options
classification	Content classification	Custom categories, confidence scoring

💾 Storage Agents

Data persistence and retrieval from various storage systems:

Agent Type	Purpose	Key Features
csv_reader	Read CSV files	Collection-based access, format options
csv_writer	Write CSV data	Append/overwrite modes, validation
json_reader	Read JSON documents	Path-based access, collection support
json_writer	Write JSON data	Document management, nested paths

📁 File Agents

File system operations for document processing:

Agent Type	Purpose	Key Features
file_reader	Read file contents	Multiple formats, encoding detection
file_writer	Write file contents	Path creation, backup options
directory_list	List directory contents	Filtering, recursive scanning

🔧 Control Flow Agents

Workflow orchestration and routing logic:

Agent Type	Purpose	Key Features
orchestrator	Intelligent routing	Multi-strategy node selection
router	Conditional routing	Rule-based decision making
parallel	Parallel execution	Concurrent processing, synchronization

🏗️ Core Agents

Basic building blocks for data flow and workflow control:

Agent Type	Purpose	Key Features
input	User input collection	Interactive prompts, validation
echo	Data passthrough	State inspection, debugging
default	Custom processing	Placeholder for custom logic

Interactive Agent Catalog

Explore all available agents with examples and configuration options:

AgentMap Agent Catalog

Browse all available agent types with examples and configurations

⚙️ Core

DefaultAgent

default

The simplest agent that logs its execution and returns a message with the prompt.

Input Fields: Any (unused)

Output Field: Returns a message including the agent's prompt

Prompt Usage: Included in output message

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,Start,,Basic node,default,Next,,input,output,Hello World

⚙️ Core

EchoAgent

echo

Simply returns the input data unchanged.

Input Fields: Returns the first input field it finds

Output Field: The input data unchanged

Prompt Usage: Ignored

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,Echo,,Echo node,echo,Next,,message,response,

⚙️ Core

BranchingAgent

branching

Used for testing conditional routing. Checks for success/failure indicators in inputs.

Input Fields: Looks for success, should_succeed, succeed, or branch fields

Output Field: Message describing the branching decision

Prompt Usage: Included in output message

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,Branch,,Decision point,branching,SuccessPath,FailurePath,input,decision,Make a choice

⚙️ Core

SuccessAgent

success

Testing agent that always succeeds.

Input Fields: Any (unused)

Output Field: Confirmation message

Prompt Usage: Included in output message

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,AlwaysSucceed,,Success node,success,Next,,input,result,I always succeed

⚙️ Core

FailureAgent

failure

Testing agent that always fails.

Input Fields: Any (unused)

Output Field: Confirmation message

Prompt Usage: Included in output message

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,AlwaysFail,,Failure node,failure,Next,,input,result,I always fail

⚙️ Core

InputAgent

input

Prompts for user input during execution.

Input Fields: Any (unused)

Output Field: User's input response

Prompt Usage: Shown to user as input prompt

Protocols:BaseAgent

Services:None required

CSV Example:

TestGraph,GetInput,,User input node,input,Process,,message,user_input,Please enter your name:

🧠 LLM

LLMAgent

llm

Uses configurable LLM providers for text generation with intelligent routing support.

Input Fields: Used to format the prompt template

Output Field: LLM response

Prompt Usage: Used as prompt template or system message

Protocols:LLMCapableAgentPromptCapableAgent

Services:LLMService

Context Options:routing_enabledtask_typeprovidermodeltemperaturememory_key

CSV Example:

QAGraph,Question,{"routing_enabled": true, "task_type": "analysis"},Ask a question,llm,Answer,,question,response,Answer this question: {question}

🧠 LLM

OpenAIAgent

openai(aliases: gptchatgpt)

Backward compatibility wrapper for LLMAgent with OpenAI provider.

Input Fields: Used to format the prompt template

Output Field: LLM response

Prompt Usage: Used as prompt template

Protocols:LLMCapableAgent

Services:LLMService (OpenAI)

Context Options:modeltemperaturemax_tokens

CSV Example:

QAGraph,Question,{"model": "gpt-4", "temperature": 0.7},Ask a question,openai,Answer,,question,response,Answer this question: {question}

🧠 LLM

AnthropicAgent

claude(aliases: claude)

Backward compatibility wrapper for LLMAgent with Anthropic provider.

Input Fields: Used to format the prompt template

Output Field: LLM response

Prompt Usage: Used as prompt template

Protocols:LLMCapableAgent

Services:LLMService (Anthropic)

Context Options:modeltemperaturemax_tokens

CSV Example:

QAGraph,Summarize,{"model": "claude-3-5-sonnet-20241022"},Summarize text,claude,Next,,text,summary,Summarize this text in 3 bullet points: {text}

🧠 LLM

GoogleAgent

gemini(aliases: gemini)

Backward compatibility wrapper for LLMAgent with Google provider.

Input Fields: Used to format the prompt template

Output Field: LLM response

Prompt Usage: Used as prompt template

Protocols:LLMCapableAgent

Services:LLMService (Google)

Context Options:modeltemperaturemax_tokens

CSV Example:

QAGraph,Generate,{"model": "gemini-1.0-pro"},Generate content,gemini,Next,,prompt,content,Generate content based on: {prompt}

💾 Storage

CSVReaderAgent

csv_reader

Read from CSV files using the unified storage system.

Input Fields: Must contain collection (file path), optional document_id, query, path

Output Field: CSV data

Prompt Usage: Optional CSV path override

Protocols:StorageCapableAgentCSVCapableAgent

Services:StorageService

Context Options:formatid_fieldencodingdelimiter

CSV Example:

DataGraph,ReadCustomers,{"format": "records", "id_field": "customer_id"},Read customer data,csv_reader,Process,,collection,customers,data/customers.csv

💾 Storage

CSVWriterAgent

csv_writer

Write to CSV files using the unified storage system.

Input Fields: Must contain data and collection (file path)

Output Field: Operation result

Prompt Usage: Optional CSV path override

Protocols:StorageCapableAgentCSVCapableAgent

Services:StorageService

Context Options:formatmodeencodingdelimiter

CSV Example:

DataGraph,WriteResults,{"format": "records", "mode": "write"},Write processed data,csv_writer,End,,data,result,data/output.csv

💾 Storage

JSONDocumentReaderAgent

json_reader

Read from JSON files using the unified storage system.

Input Fields: Must contain collection (file path), optional document_id, path

Output Field: JSON data

Prompt Usage: Optional JSON path override

Protocols:StorageCapableAgentJSONCapableAgent

Services:StorageService

Context Options:formatencoding

CSV Example:

ConfigGraph,ReadConfig,{"format": "dict", "encoding": "utf-8"},Read configuration,json_reader,Process,,collection,config,config/app.json

💾 Storage

JSONDocumentWriterAgent

json_writer

Write to JSON files using the unified storage system.

Input Fields: Must contain data and collection (file path)

Output Field: Operation result

Prompt Usage: Optional JSON path override

Protocols:StorageCapableAgentJSONCapableAgent

Services:StorageService

Context Options:formatindentencoding

CSV Example:

ConfigGraph,SaveState,{"format": "dict", "indent": 2},Save application state,json_writer,End,,state,result,data/state.json

💾 Storage

VectorReaderAgent

vector_reader

Work with vector databases for semantic search and document retrieval.

Input Fields: query for similarity search

Output Field: Retrieved documents

Prompt Usage: Optional configuration override

Protocols:StorageCapableAgentVectorCapableAgent

Services:StorageService

Context Options:providerembedding_modelsimilarity_thresholdmax_results

CSV Example:

VectorGraph,Search,{"similarity_threshold": 0.8, "max_results": 5},Search for similar documents,vector_reader,Process,,query,search_results,

💾 Storage

VectorWriterAgent

vector_writer

Work with vector databases for embedding and storing documents.

Input Fields: document data

Output Field: Operation status

Prompt Usage: Optional configuration override

Protocols:StorageCapableAgentVectorCapableAgent

Services:StorageService

Context Options:providerembedding_modelcollection_name

CSV Example:

VectorGraph,LoadDocs,{"provider": "chroma", "embedding_model": "text-embedding-ada-002"},Load documents into vector store,vector_writer,Search,,documents,load_result,

📁 File

FileReaderAgent

file_reader

Reads and processes various document types with optional chunking and filtering.

Input Fields: collection (file path), optional document_id, query, path, format

Output Field: Document data with metadata

Prompt Usage: Not used

Protocols:StorageCapableAgentFileCapableAgent

Services:StorageService

Context Options:chunk_sizechunk_overlapshould_splitinclude_metadata

CSV Example:

DocGraph,ReadDocs,{"chunk_size": 1000, "chunk_overlap": 200, "should_split": true},Read documents,file_reader,Process,,collection,documents,

📁 File

FileWriterAgent

file_writer

Writes content to various text-based formats with different write modes.

Input Fields: collection (file path), data (content), optional mode

Output Field: Write operation result

Prompt Usage: Not used

Protocols:StorageCapableAgentFileCapableAgent

Services:StorageService

Context Options:modeencoding

CSV Example:

DocGraph,WriteFile,{"mode": "write", "encoding": "utf-8"},Write document,file_writer,Next,,data,result,path/to/output.txt

🔧 Specialized

OrchestrationAgent

orchestrator

Routes execution to one or more nodes based on context configuration.

Input Fields: available_nodes and routing data

Output Field: selected nodes

Prompt Usage: Not used

Protocols:BaseAgent

Services:None required

Context Options:nodes

CSV Example:

WorkflowGraph,Router,{"nodes": "ProcessA|ProcessB|ProcessC"},Route to processors,orchestrator,Collect,Error,available_nodes|data,selected_nodes,

🔧 Specialized

SummaryAgent

summary

Combines multiple inputs into a structured summary.

Input Fields: Multiple input fields to combine

Output Field: Combined summary result

Prompt Usage: Not used

Protocols:BaseAgent

Services:None required

Context Options:formatinclude_keys

CSV Example:

DataGraph,Combine,{"format": "{key}: {value}\n"},Combine results,summary,Next,Error,result_a|result_b|result_c,combined,

Found 20 of 20 agents. Need help? Check out the detailed agent documentation or quick start guide.

Agent Implementation Examples

LLM Agent with Process Method

class LLMAgent(BaseAgent, LLMCapableAgent):
    """LLM agent demonstrating process() method implementation"""
    
    def configure_llm_service(self, llm_service: LLMServiceProtocol):
        self._llm_service = llm_service
    
    def process(self, inputs: Dict[str, Any]) -> Any:
        # Extract input data
        user_query = inputs.get('query', '')
        context = inputs.get('context', '')
        
        # Build messages for LLM
        messages = [
            {"role": "system", "content": self.prompt},
            {"role": "user", "content": f"Query: {user_query}\nContext: {context}"}
        ]
        
        # Call LLM service
        response = self.llm_service.call_llm(
            provider="anthropic",
            messages=messages,
            model="claude-3-5-sonnet-20241022"
        )
        
        return response

Storage Agent with Process Method

class CSVReaderAgent(BaseAgent, StorageCapableAgent):
    """CSV reader agent with storage service injection"""
    
    def configure_storage_service(self, storage_service: StorageServiceProtocol):
        self._storage_service = storage_service
    
    def process(self, inputs: Dict[str, Any]) -> Any:
        # Extract input parameters
        collection = inputs.get('collection', 'default')
        document_id = inputs.get('document_id')
        format_type = self.context.get('format', 'records')
        
        # Read data using storage service
        data = self.storage_service.read(
            collection=collection,
            document_id=document_id,
            format=format_type
        )
        
        return data

Custom Agent Example

class WeatherAgent(BaseAgent):
    """Custom weather agent showing process() method pattern"""
    
    def process(self, inputs: Dict[str, Any]) -> Any:
        location = inputs.get('location', 'Unknown')
        
        # Simulate weather API call
        weather_data = {
            'location': location,
            'temperature': '72°F',
            'conditions': 'Sunny',
            'humidity': '45%'
        }
        
        # Format response based on prompt
        if 'brief' in self.prompt.lower():
            return f"{weather_data['conditions']}, {weather_data['temperature']}"
        else:
            return weather_data

CSV Configuration Patterns

Basic Agent Configuration

workflow,node,description,type,next_node,error_node,input_fields,output_field,prompt
MyFlow,Process,Process user input,llm,End,Error,user_input,ai_response,Answer this question: {user_input}

Agent with Context Configuration

workflow,node,description,type,next_node,error_node,input_fields,output_field,prompt,context
MyFlow,Classify,Classify content,llm,Route,Error,content,category,Classify this content,"{""provider"": ""openai"", ""temperature"": 0.1}"

Memory-Enabled Agent

workflow,node,description,type,next_node,error_node,input_fields,output_field,prompt,context
ChatBot,Respond,Generate response,llm,Listen,Error,user_input|conversation,response,You are a helpful assistant,"{""memory_key"": ""conversation"", ""max_memory_messages"": 10}"

Service Integration

Protocol-Based Service Injection

Agents that need business services implement specific protocols:

# LLM-capable agents
class MyAgent(BaseAgent, LLMCapableAgent):
    def configure_llm_service(self, llm_service: LLMServiceProtocol):
        self._llm_service = llm_service

# Storage-capable agents  
class MyAgent(BaseAgent, StorageCapableAgent):
    def configure_storage_service(self, storage_service: StorageServiceProtocol):
        self._storage_service = storage_service

# Multiple capabilities
class MyAgent(BaseAgent, LLMCapableAgent, StorageCapableAgent):
    def configure_llm_service(self, llm_service: LLMServiceProtocol):
        self._llm_service = llm_service
    
    def configure_storage_service(self, storage_service: StorageServiceProtocol):
        self._storage_service = storage_service

Infrastructure Service Access

All agents have access to infrastructure services:

def process(self, inputs: Dict[str, Any]) -> Any:
    # Logging
    self.logger.info(f"Processing inputs: {list(inputs.keys())}")
    
    # Execution tracking
    self.execution_tracker_service.record_node_start(self.name, inputs)
    
    # State operations
    current_state = self.state_adapter_service.get_current_state()
    
    # Your processing logic
    result = self.custom_processing(inputs)
    
    return result

Error Handling Patterns

Built-in Error Handling

def process(self, inputs: Dict[str, Any]) -> Any:
    try:
        # Processing logic
        return self.do_processing(inputs)
    except Exception as e:
        # Errors are automatically caught and routed to error_node
        self.logger.error(f"Processing failed: {e}")
        raise  # Re-raise for proper error routing

Graceful Degradation

def process(self, inputs: Dict[str, Any]) -> Any:
    # Try primary method
    if self._llm_service:
        try:
            return self._llm_service.call_llm(messages=messages)
        except Exception as e:
            self.logger.warning(f"LLM failed, using fallback: {e}")
    
    # Fallback method
    return self.simple_text_processing(inputs)

Best Practices

Agent Design

1. Single Purpose: Each agent should have one clear responsibility
2. Stateless Processing: Avoid storing state in agent instances
3. Error Resilient: Handle exceptions gracefully with clear error messages
4. Service Optional: Design agents to work with or without optional services

Process Method Guidelines

1. Input Validation: Validate required inputs at the start
2. Clear Returns: Return simple values that can be JSON serialized
3. Logging: Log key processing steps for debugging
4. Documentation: Include docstrings explaining input/output contracts

Configuration Patterns

1. Use Context: Store configuration in the context field
2. Default Values: Provide sensible defaults for optional parameters
3. Input Fields: Explicitly declare required input fields
4. Output Field: Always specify the output field name

Next Steps

• Custom Agent Development - Build your own agent types