# LionAGI Action System Guide
## Overview
The Action System provides core functionality for managing and executing tools within LionAGI. It implements a type-safe, observable framework for:
- Tool registration and execution
- Function schema generation
- Execution state tracking
- Resource lifecycle management
- Processing pipelines
> [!info] Core Purpose
> The system is designed to provide controlled execution of functions with comprehensive monitoring, validation, and resource management.
## System Architecture
### Component Hierarchy
```mermaid
classDiagram
Element <|-- Tool
Element <|-- ObservableAction
ObservableAction <|-- FunctionCalling
class Tool {
+function: Callable
+schema_: dict
+pre_processor: Callable
+post_processor: Callable
}
class ObservableAction {
+status: EventStatus
+execution_time: float
+to_log()
}
class FunctionCalling {
+func_tool: Tool
+arguments: dict
+invoke()
}
```
### Core Components
1. **Tool**
```python
class Tool(Element):
"""
Wraps a function with processing capabilities.
"""
function: Callable[..., Any]
schema_: dict[str, Any] | None
pre_processor: Callable[[Any], Any] | None
post_processor: Callable[[Any], Any] | None
parser: Callable[[Any], Any] | None
```
2. **ObservableAction**
```python
class ObservableAction(Element):
"""
Base for actions with execution tracking.
"""
status: EventStatus
execution_time: float | None
execution_response: Any
execution_error: str | None
```
3. **FunctionCalling**
```python
class FunctionCalling(ObservableAction):
"""
Handles function execution pipeline.
"""
func_tool: Tool | None
arguments: dict[str, Any] | None
function: str | None
```
### Event States
```python
class EventStatus(str, Enum):
"""Track action execution progress."""
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
```
## Implementation Guide
### 1. Tool Creation
The basic pattern for creating tools:
```python
from lionagi.core.action import tool, Tool
# Using decorator
@tool
async def process_data(data: dict) -> dict:
"""Process input data.
Args:
data: Input dictionary
Returns:
dict: Processed result
"""
return await process(data)
# Direct creation
processor = Tool(
function=process_function,
pre_processor=validate_input,
post_processor=format_output
)
```
> [!note] Schema Generation
> Tools automatically generate OpenAI function schemas from:
> - Function signatures
> - Type annotations
> - Docstrings
> - Default values
### 2. Action Manager Usage
```python
from lionagi.core.action import ActionManager
class ProcessingManager:
def __init__(self):
self.manager = ActionManager()
def setup(self):
# Register tools
self.manager.register_tools([
process_data,
validate_data
])
async def process(self, data: dict):
# Execute function
result = await self.manager.invoke({
"function": "process_data",
"arguments": {"data": data}
})
return result
```
> [!important] Tool Registration
> - Tools must have unique names
> - Functions must be callable
> - Functions must have names
> - Schema is auto-generated unless provided
### 3. Processing Pipeline
The execution flow in the system:
```mermaid
sequenceDiagram
participant Client
participant ActionManager
participant Tool
participant Function
Client->>ActionManager: invoke()
ActionManager->>Tool: match_tool()
Tool->>Function: pre_process()
Function->>Function: execute()
Function->>Tool: post_process()
Tool->>ActionManager: return result
ActionManager->>Client: return processed
```
### 4. Event Tracking
```python
async def execute_with_tracking(action: ObservableAction):
try:
# Start execution
action.status = EventStatus.PROCESSING
# Execute action
result = await action.execute()
action.status = EventStatus.COMPLETED
return result
except Exception as e:
action.status = EventStatus.FAILED
action.execution_error = str(e)
raise
```
## Best Practices
### 1. Tool Design
> [!tip] Tool Guidelines
> - Keep functions focused and single-purpose
> - Use comprehensive type hints
> - Write clear docstrings for schema generation
> - Keep processing pipeline steps simple
> - Validate inputs and outputs
### 2. Error Handling
```python
async def safe_execute(manager: ActionManager, func_call: dict):
try:
return await manager.invoke(func_call)
except ValueError as e:
# Handle invalid function or arguments
logger.error(f"Invalid function call: {e}")
raise
except Exception as e:
# Handle execution errors
logger.error(f"Execution failed: {e}")
raise
```
### 3. Resource Usage
> [!warning] Resource Management
> - Track execution times
> - Log function calls
> - Monitor resource usage
> - Clean up after execution
> - Use proper error boundaries
## Common Patterns
### 1. Function Pipeline
```python
def create_pipeline(
function: Callable,
validators: list[Callable],
formatters: list[Callable]
) -> Tool:
"""Create a tool with processing pipeline."""
async def pre_process(**kwargs):
# Run all validators
for validator in validators:
kwargs = await validator(**kwargs)
return kwargs
async def post_process(result):
# Apply all formatters
for formatter in formatters:
result = await formatter(result)
return result
return Tool(
function=function,
pre_processor=pre_process,
post_processor=post_process
)
```
### 2. Batch Processing
```python
async def batch_process(
manager: ActionManager,
items: list[dict]
) -> list:
"""Process items in batches."""
results = []
for item in items:
result = await manager.invoke({
"function": "process_item",
"arguments": item
})
results.append(result)
return results
```
## Integration Examples
### 1. With Branch System
```python
from lionagi.core.branch import Branch
class ActionBranch(Branch):
def __init__(self):
self.action_manager = ActionManager()
async def process_action(self, action_request):
# Convert to function call
func_call = {
"function": action_request.function,
"arguments": action_request.arguments
}
# Execute and track
result = await self.action_manager.invoke(func_call)
return result
```
### 2. With Communication System
```python
from lionagi.core.communication import MessageManager
class ActionProcessor:
def __init__(self):
self.action_manager = ActionManager()
self.message_manager = MessageManager()
async def process_message(self, message):
if message.has_action:
result = await self.action_manager.invoke(
message.get_action()
)
return await self.message_manager.create_response(
content=result
)
```
## See Also
- [[Action System|Action System API Reference]]