Service Reference
Quick reference for all AgentMap services, organized by category with context parameters and implementation details.
AgentMap Service Catalog
Browse all available services with context parameters and usage examples
GraphBuilderService
Builds graph models from CSV files and pandas DataFrames with comprehensive validation.
graph_builder = container.graph_builder_service()
graph = graph_builder.build_from_csv(Path("workflow.csv"))CompilationService
Compiles graph models into executable LangGraph StateGraphs with state schema support.
compilation = container.compilation_service()
compiled = compilation.compile_graph("MyWorkflow", Path("workflow.csv"))GraphRunnerService
Orchestrates graph execution with comprehensive tracking, policy evaluation, and state management.
runner = container.graph_runner_service()
result = runner.run_graph("MyWorkflow", {"input": "data"})
print(f"Success: {result.success}, Duration: {result.duration}s")OrchestratorService
Provides intelligent node selection and workflow routing with multiple matching strategies.
orchestrator = container.orchestrator_service()
selected = orchestrator.select_best_node(
"I need to process a payment",
nodes_dict, strategy="tiered", confidence_threshold=0.8
)AgentFactoryService
Creates and configures agent instances with automatic dependency injection based on protocols.
factory = container.agent_factory_service() agent = factory.create_agent(node) # Agent has all required services automatically injected
GraphAssemblyService
Assembles graph models into LangGraph StateGraphs with agent integration and state schema support.
assembly = container.graph_assembly_service() state_graph = assembly.assemble_graph(graph_model, StateSchema) compiled = state_graph.compile()
LoggingService
Provides structured logging throughout the application with class-specific and agent-specific loggers.
class MyService:
def __init__(self, logging_service: LoggingService):
self.logger = logging_service.get_class_logger(self)
self.logger.info("Service initialized")StateAdapterService
Adapts state between different formats including dictionaries, Pydantic models, and custom schemas.
adapter = container.state_adapter_service()
adapted = adapter.adapt_initial_state({"input": "data"}, StateSchema)FunctionResolutionService
Resolves function references for dynamic routing and custom workflow logic.
resolver = container.function_resolution_service()
router_func = resolver.resolve_function("func:custom_router")AppConfigService
Manages application configuration with intelligent defaults and environment-based overrides.
config = container.app_config_service()
csv_path = config.get_csv_path()
openai_config = config.get_llm_config("openai")StorageConfigService
Manages storage service configuration with provider-specific settings and defaults.
storage_config = container.storage_config_service()
csv_config = storage_config.get_provider_config("csv", "local")StorageManager
Unified interface managing all storage services with automatic provider selection and registration.
storage_manager = container.storage_manager()
csv_service = storage_manager.get_service("csv")
json_service = storage_manager.get_service("json", "cloud")CSVStorageService
High-performance CSV file operations with pandas integration and flexible data format support.
csv_service = storage_manager.get_service("csv")
data = csv_service.read("users", format="records")
result = csv_service.write("users", new_data, mode="append")JSONStorageService
JSON document storage with path-based queries and structured data management.
json_service = storage_manager.get_service("json")
doc = json_service.read("configs", "app_config")
result = json_service.write("configs", {"debug": True}, "app_config")ValidationService
Comprehensive validation orchestration for CSV files, configurations, and data integrity.
validator = container.validation_service()
result = validator.validate_csv(Path("workflow.csv"))
if not result.is_valid:
for error in result.errors:
print(f"Error: {error.message}")ExecutionTrackingService
Comprehensive workflow execution tracking with metrics, history, and performance analysis.
tracking = container.execution_tracking_service()
tracker = tracking.create_tracker("MyWorkflow")
tracker.start()
# ... execution ...
tracker.complete(final_state)
summary = tracker.get_summary()ExecutionPolicyService
Configurable execution success evaluation with customizable policies and criteria.
policy = container.execution_policy_service()
success = policy.evaluate_success(execution_summary, "MyWorkflow")
print(f"Workflow success: {success}")Service Categories
Core Services
Essential business logic services that form the backbone of AgentMap workflows.
Infrastructure Services
Technical services providing logging, state management, and system utilities.
Configuration Services
Services managing application and storage configuration with smart defaults.
Storage Services
Data persistence services supporting multiple storage types and formats.
Validation Services
Services ensuring data integrity and configuration correctness.
Execution Services
Services tracking and managing workflow execution with policy evaluation.
Context Parameter Usage
All services accept context parameters in Python dictionary format:
context = {
'provider': 'openai',
'model': 'gpt-4',
'temperature': 0.7
}
When using services in CSV workflows, context should be specified as JSON:
name,description,type,context
llm_node,Generate response,llm,"{\"provider\": \"openai\", \"model\": \"gpt-4\"}"
Service Integration
Services integrate through dependency injection in the DI container:
# Services are automatically injected based on protocols
container = Container()
graph_runner = container.graph_runner_service()
# GraphRunnerService has CompilationService, ExecutionTrackingService, etc. injected
Protocol Implementation
Services implement specific protocols to define their capabilities:
- LLMServiceProtocol: Language model integration
- StorageServiceProtocol: Data persistence operations
- LoggingServiceProtocol: Structured logging functionality
- ValidationServiceProtocol: Data validation operations
Common Usage Patterns
1. Graph Building and Execution
# Build graph from CSV
graph_builder = container.graph_builder_service()
graph = graph_builder.build_from_csv(Path("workflow.csv"))
# Run with tracking
runner = container.graph_runner_service()
result = runner.run_graph("MyWorkflow", {'input': 'data'})
2. Storage Operations
# Get storage service
storage_manager = container.storage_manager()
csv_service = storage_manager.get_service("csv")
# Read/write data
data = csv_service.read("users", format="records")
result = csv_service.write("users", new_data)
3. Configuration Management
# Get configuration
config = container.app_config_service()
csv_path = config.get_csv_path()
llm_config = config.get_llm_config("openai")
4. Validation
# Validate CSV files
validator = container.validation_service()
result = validator.validate_csv(Path("workflow.csv"))
if not result.is_valid:
for error in result.errors:
print(f"Error: {error.message}")
Error Handling
Services follow consistent error handling patterns:
- Graceful Degradation: Optional services return None if unavailable
- Clear Error Messages: ValidationResult objects with detailed error information
- Logging Integration: All errors are properly logged with context
- Type Safety: Strong typing prevents common configuration errors
Service Lifecycle
- Lazy Creation: Services created when first requested from container
- Singleton Pattern: Services cached for reuse within container scope
- Dependency Injection: All dependencies automatically resolved
- Optional Services: LLM and external services handle missing configuration gracefully
Best Practices
- Always use DI container - Never instantiate services directly
- Check for None returns - Optional services may not be available
- Use proper typing - Leverage type hints for better IDE support
- Handle validation errors - Always check ValidationResult objects
- Use context parameters - Configure services through context when possible
Adding Custom Services
To extend AgentMap with custom services:
- Create service class with proper constructor injection
- Implement relevant protocols for capability declaration
- Register in DI container with dependency resolution
- Add to documentation following the patterns shown above
- Write comprehensive tests using the real DI container
Example custom service:
class CustomAnalyticsService:
def __init__(self, storage_manager: StorageManager,
logging_service: LoggingService):
self.storage = storage_manager
self.logger = logging_service.get_class_logger(self)
def track_event(self, event: str, data: Dict[str, Any]) -> bool:
self.logger.info(f"Tracking event: {event}")
# Implementation here
return True
# Register in container
class Container:
def custom_analytics_service(self) -> CustomAnalyticsService:
return self._get_or_create('_custom_analytics_service',
lambda: CustomAnalyticsService(
storage_manager=self.storage_manager(),
logging_service=self.logging_service()
)
)