| Name | aitor JSON |
| Version |
1.1.2
JSON |
| download |
| home_page | None |
| Summary | A Python framework for building intelligent, memory-enabled agents with ReAct reasoning |
| upload_time | 2025-07-23 10:50:17 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.9 |
| license | MIT |
| keywords |
ai
workflow
builder
react
agents
llm
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Aitor
Aitor is a comprehensive Python framework for building intelligent, memory-enabled agents that can execute complex workflows with advanced reasoning capabilities. The framework combines DAG-based workflow execution with ReAct (Reasoning and Acting) agents, planning agents, and LLM integration to create powerful AI-driven applications.
## 🚀 Features
### Core Framework
- **Memory-Enabled Agents**: Stateful agents with typed memory management and persistence
- **DAG Workflows**: Task dependency management through directed acyclic graphs
- **Async Processing**: Thread-safe execution with both blocking (`ask`) and non-blocking (`tell`) APIs
- **Task Chaining**: Intuitive `>>` operator for defining task dependencies
### AI Agent Types
- **ReAct Agents**: Reasoning and Acting agents with Think → Act → Observe loops
- **Planning Agents**: Advanced agents that break down complex tasks into manageable todos
- **Sub-Agent Management**: Delegate specialized tasks to focused sub-agents
- **LLM Integration**: Support for OpenAI, Anthropic, and custom LLM providers
### Advanced Capabilities
- **Tool Registry**: Dynamic tool management with async execution
- **Structured Responses**: JSON-based responses using Pydantic models
- **Memory Persistence**: Export/import memory for session management
- **Workflow Visualization**: Built-in Mermaid diagram generation
## 📦 Installation
```bash
pip install aitor
```
For development:
```bash
git clone https://github.com/Ashfakh/aitor.git
cd aitor
pip install -e .
```
## 🏃 Quick Start
### Basic Workflow Agent
```python
import asyncio
from typing import List
from aitor import Aitor, Aitorflow, task
@task
def clean_text(text: str) -> str:
return text.strip().lower()
@task
def count_words(text: str) -> int:
return len(text.split())
@task
def analyze_text(text: str, word_count: int) -> dict:
return {
"text": text,
"word_count": word_count,
"avg_word_length": len(text.replace(" ", "")) / word_count if word_count > 0 else 0
}
async def text_handler(message: str, aitor: Aitor[List[str]]):
# Store in memory
memory = aitor.get_memory()
memory.append(message)
aitor.set_memory(memory)
# Execute workflow if attached
if aitor.workflow:
return await asyncio.to_thread(aitor.workflow.execute, message)
return f"Processed: {message}"
async def main():
# Create workflow
workflow = Aitorflow(name="Text Analysis")
# Define task dependencies
clean_text >> count_words >> analyze_text
clean_text >> analyze_text # Multiple dependencies
workflow.add_task(clean_text)
# Create agent
aitor = Aitor(
initial_memory=[],
name="TextProcessor",
on_receive_handler=text_handler
)
aitor.attach_workflow(workflow)
# Process text
result = await aitor.ask(" Hello World Example! ")
print(f"Result: {result}")
Aitor.shutdown()
if __name__ == "__main__":
asyncio.run(main())
```
### ReAct Agent with Tools
```python
import asyncio
from aitor import create_react_agent
from aitor.tools import tool
@tool(name="calculator", description="Perform mathematical calculations")
def calculate(expression: str) -> float:
"""Safely evaluate mathematical expressions."""
try:
# Simple validation for safety
allowed_chars = set('0123456789+-*/()., ')
if not all(c in allowed_chars for c in expression):
raise ValueError("Invalid characters in expression")
return eval(expression)
except Exception as e:
raise ValueError(f"Calculation error: {e}")
@tool(name="text_analyzer", description="Analyze text properties")
def analyze_text(text: str) -> dict:
"""Analyze various properties of text."""
return {
"length": len(text),
"word_count": len(text.split()),
"sentence_count": text.count('.') + text.count('!') + text.count('?'),
"uppercase_ratio": sum(1 for c in text if c.isupper()) / len(text) if text else 0
}
async def main():
# Create ReAct agent
agent = await create_react_agent(
name="MathTextAgent",
llm_provider="openai",
llm_config={
"api_key": "your-openai-api-key",
"model": "gpt-4"
},
tools=[calculate, analyze_text],
max_reasoning_steps=10
)
# Solve complex problems
response = await agent.solve(
"Calculate the square root of 144, then analyze the text 'Hello World!' "
"and tell me the relationship between the calculation result and word count."
)
print(f"Agent Response: {response}")
# Export memory for persistence
memory_data = agent.export_memory()
print(f"Conversation history: {len(memory_data['conversation_history'])} messages")
await agent.shutdown()
if __name__ == "__main__":
asyncio.run(main())
```
### Planning Agent with Todos
```python
import asyncio
from aitor import PlanningReactAgent
from aitor.llm import LLMManager
async def main():
# Setup LLM
llm_manager = LLMManager()
llm_manager.add_provider(
name="openai",
provider="openai",
config={"api_key": "your-api-key", "model": "gpt-4"}
)
# Create planning agent
agent = PlanningReactAgent(
name="ProjectPlanner",
llm_manager=llm_manager,
max_reasoning_steps=20
)
# Complex planning task
response = await agent.solve(
"Help me plan and execute a data analysis project. I need to collect data "
"from APIs, clean it, perform statistical analysis, and create visualizations."
)
print(f"Planning Response: {response}")
# Check created todos
memory = agent.get_memory()
print(f"Created {len(memory.todos)} todos:")
for todo in memory.todos:
print(f" - [{todo.status}] {todo.title}")
await agent.shutdown()
if __name__ == "__main__":
asyncio.run(main())
```
## 🧠 Agent Types
### 1. Base Aitor Agent
Memory-enabled agents with workflow integration:
- Generic typed memory: `Aitor[T]`
- Thread-safe operations
- Workflow attachment
- Async processing
### 2. ReAct Agents
Reasoning and Acting agents that follow Think → Act → Observe loops:
- **Tool Integration**: Dynamic tool registry with validation
- **Reasoning Engine**: Step-by-step problem solving
- **Memory Management**: Conversation and reasoning history
- **LLM Integration**: Support for multiple providers
### 3. Planning Agents
Advanced agents that break complex tasks into manageable todos:
- **Todo Management**: Create, track, and execute todos with priorities
- **Sub-Agent Delegation**: Spawn specialized agents for specific tasks
- **Plan Execution**: Systematic approach to complex problems
- **Progress Tracking**: Monitor todo completion and overall progress
## 🛠️ Core Components
### Memory System
```python
from aitor.memory import ReactMemory
# Structured memory with conversation, tools, and reasoning
memory = ReactMemory()
memory.add_message("user", "Hello!")
memory.add_tool_execution("calculator", {"expression": "2+2"}, 4)
memory.add_reasoning_step("THINK", "I need to solve this math problem")
```
### Tool Registry
```python
from aitor.tools import ToolRegistry, tool
registry = ToolRegistry()
@tool(name="example", description="An example tool")
def example_tool(param: str) -> str:
return f"Processed: {param}"
await registry.register_tool(example_tool)
result = await registry.execute_tool("example", {"param": "test"})
```
### LLM Management
```python
from aitor.llm import LLMManager
llm_manager = LLMManager()
# Add multiple providers
llm_manager.add_provider("openai", "openai", {
"api_key": "...",
"model": "gpt-4"
})
llm_manager.add_provider("claude", "anthropic", {
"api_key": "...",
"model": "claude-3-opus-20240229"
})
# Switch between providers
llm_manager.set_default_provider("claude")
```
### Workflow Visualization
```python
from aitor import Aitorflow
workflow = Aitorflow(name="Example")
# ... add tasks ...
# Generate Mermaid diagram
mermaid_code = workflow.visualize()
print(mermaid_code)
```
## 📚 Advanced Usage
### Tool Creation Patterns
#### 1. Basic Tool Creation
```python
from aitor.tools import tool
@tool(
name="calculator",
description="Perform mathematical calculations",
timeout=10.0,
async_execution=True
)
def calculate(expression: str) -> float:
"""Calculate the result of a mathematical expression."""
return eval(expression)
```
#### 2. Tools with Pre-configured Parameters
For scenarios where you need to pass static parameters (like tenant details) that the agent doesn't provide:
```python
from aitor.tools import tool
def create_database_tool(tenant_id: str, connection_string: str):
"""Factory function to create tenant-specific database tool."""
@tool(
name=f"database_query_{tenant_id}",
description=f"Query database for tenant {tenant_id}"
)
def query_database(query: str) -> dict:
"""Execute database query with pre-configured tenant."""
# tenant_id and connection_string are captured from closure
return execute_query(connection_string, tenant_id, query)
return query_database
# Usage during agent creation
tenant_db_tool = create_database_tool("tenant_123", "postgresql://...")
```
#### 3. Using functools.partial for Parameter Binding
```python
from functools import partial
from aitor.tools import tool
def database_operation(tenant_id: str, api_key: str, query: str) -> dict:
"""Base database function with all parameters."""
return {
"tenant": tenant_id,
"query": query,
"result": f"Query executed for {tenant_id}"
}
# Create tenant-specific tool using partial
tenant_specific_db = partial(
database_operation,
tenant_id="tenant_123",
api_key="secret_key"
)
@tool(name="tenant_database", description="Query database for current tenant")
def tenant_db_tool(query: str) -> dict:
return tenant_specific_db(query=query)
```
#### 4. Custom Tool Class with Context
```python
from aitor.tools import Tool
from typing import Any, Dict
class TenantTool:
"""Tool with pre-configured tenant parameters."""
def __init__(self, tenant_id: str, api_key: str, base_url: str):
self.tenant_id = tenant_id
self.api_key = api_key
self.base_url = base_url
def search_documents(self, query: str, limit: int = 10) -> dict:
"""Search documents for this tenant."""
return {
"tenant": self.tenant_id,
"query": query,
"limit": limit,
"results": f"Documents for {self.tenant_id}"
}
def create_tool(self) -> Tool:
"""Create the tool instance."""
return Tool(
name=f"search_docs_{self.tenant_id}",
func=self.search_documents,
description=f"Search documents for tenant {self.tenant_id}",
timeout=30.0
)
# Usage
tenant_tool_factory = TenantTool("tenant_123", "api_key", "https://api.example.com")
search_tool = tenant_tool_factory.create_tool()
```
#### 5. Environment/Context-Based Configuration
```python
from aitor.tools import tool
from dataclasses import dataclass
@dataclass
class TenantContext:
tenant_id: str
database_url: str
api_key: str
storage_bucket: str
def create_tenant_tools(context: TenantContext):
"""Create all tools for a specific tenant context."""
@tool(name="fetch_user_data", description="Fetch user data for tenant")
def fetch_user_data(user_id: str) -> dict:
# context is captured in closure
return fetch_from_db(context.database_url, context.tenant_id, user_id)
@tool(name="upload_file", description="Upload file to tenant storage")
def upload_file(file_name: str, content: str) -> str:
return upload_to_storage(context.storage_bucket, context.tenant_id, file_name, content)
@tool(name="send_notification", description="Send notification via tenant API")
def send_notification(message: str, recipient: str) -> bool:
return send_via_api(context.api_key, context.tenant_id, message, recipient)
return [fetch_user_data, upload_file, send_notification]
# Usage
tenant_context = TenantContext(
tenant_id="tenant_123",
database_url="postgresql://...",
api_key="secret_123",
storage_bucket="tenant-123-files"
)
tools = create_tenant_tools(tenant_context)
# Register all tenant tools with agent
for tool in tools:
await agent.register_tool(tool)
```
#### 6. Complete Tenant-Specific Agent Example
```python
from aitor import create_react_agent
from aitor.tools import tool
class CustomerServiceAgent:
def __init__(self, tenant_id: str, customer_db_url: str, support_api_key: str):
self.tenant_id = tenant_id
self.customer_db_url = customer_db_url
self.support_api_key = support_api_key
def create_tools(self):
"""Create tenant-specific tools."""
@tool(name="lookup_customer", description="Look up customer information")
def lookup_customer(customer_id: str) -> dict:
return query_customer_db(self.customer_db_url, self.tenant_id, customer_id)
@tool(name="create_ticket", description="Create support ticket")
def create_ticket(title: str, description: str, priority: str = "medium") -> str:
return create_support_ticket(
self.support_api_key,
self.tenant_id,
title,
description,
priority
)
@tool(name="get_billing_info", description="Get customer billing information")
def get_billing_info(customer_id: str) -> dict:
return get_billing_data(self.customer_db_url, self.tenant_id, customer_id)
return [lookup_customer, create_ticket, get_billing_info]
async def create_agent(self, llm_manager):
"""Create the complete agent with tenant-specific tools."""
tools = self.create_tools()
agent = await create_react_agent(
name=f"CustomerServiceAgent_{self.tenant_id}",
llm_manager=llm_manager,
tools=tools,
agent_role="customer service representative",
additional_instructions=f"You are helping customers for tenant {self.tenant_id}"
)
return agent
# Usage
cs_agent_factory = CustomerServiceAgent(
tenant_id="acme_corp",
customer_db_url="postgresql://...",
support_api_key="support_key_123"
)
agent = await cs_agent_factory.create_agent(llm_manager)
```
#### Key Benefits of Pre-configured Tools:
- **Security**: Sensitive credentials are encapsulated and not exposed to the agent
- **Simplicity**: Agent only needs to pass business logic parameters
- **Tenant Isolation**: Each agent instance is pre-configured for specific tenant
- **Reusability**: Factory pattern allows creating multiple tenant-specific agents
- **Type Safety**: Pre-configured parameters are validated at creation time
### Avoiding Tool Registry Conflicts
When creating multiple agents, ensure each has its own tool registry to prevent "Tool already registered" errors:
```python
from aitor import ReactAgentBuilder
from aitor.tools import ToolRegistry
# Method 1: Explicit tool registry per agent
def create_agent_with_tools(tenant_id: str):
# Create fresh tool registry for this agent
tool_registry = ToolRegistry()
# Create tenant-specific tools
@tool(name="lookup_info", description=f"Look up info for {tenant_id}")
def lookup_info(query: str) -> dict:
return get_tenant_info(tenant_id, query)
# Build agent with explicit registry
agent = await (ReactAgentBuilder()
.name(f"Agent_{tenant_id}")
.tool_registry(tool_registry) # Explicit registry
.add_tool(lookup_info)
.build())
return agent
# Method 2: Using builder pattern (recommended)
async def create_isolated_agents():
# Each agent gets its own tool registry automatically
agent1 = await (ReactAgentBuilder()
.name("Agent1")
.add_tool(create_lookup_tool("tenant1"))
.build())
agent2 = await (ReactAgentBuilder()
.name("Agent2")
.add_tool(create_lookup_tool("tenant2")) # Same tool name, different registry
.build())
return agent1, agent2
# Method 3: Factory pattern with unique tool names
def create_unique_tools(tenant_id: str):
@tool(name=f"lookup_info_{tenant_id}", description=f"Look up info for {tenant_id}")
def lookup_info(query: str) -> dict:
return get_tenant_info(tenant_id, query)
return lookup_info
```
### Memory Persistence
```python
# Export memory
memory_data = agent.export_memory()
# Save to file
import json
with open("agent_memory.json", "w") as f:
json.dump(memory_data, f, indent=2)
# Load and import
with open("agent_memory.json", "r") as f:
memory_data = json.load(f)
new_agent.import_memory(memory_data)
```
### Sub-Agent Architecture
```python
from aitor.sub_agent import SubAgentManager
# Create specialized sub-agents
sub_manager = SubAgentManager()
sub_manager.create_sub_agent(
name="DataAnalyst",
specialization="statistical analysis and data processing",
tools=[pandas_tool, numpy_tool, matplotlib_tool]
)
# Delegate tasks
result = await sub_manager.delegate_task(
"DataAnalyst",
"Analyze this dataset and create visualizations"
)
```
## 🎯 Use Cases
- **Data Processing Pipelines**: Complex ETL workflows with error handling
- **AI-Powered Assistants**: Conversational agents with tool access
- **Automated Planning**: Break down complex projects into actionable steps
- **Research Automation**: Gather, analyze, and synthesize information
- **Code Generation**: AI agents that write and test code
- **Content Creation**: Multi-step content workflows with reviews
## 🧪 Development
### Setup Development Environment
```bash
# Clone repository
git clone https://github.com/Ashfakh/aitor.git
cd aitor
# Create virtual environment
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# Install in development mode
pip install -e .
# Install development dependencies
pip install -e .[dev]
```
### Running Tests
```bash
# Run linting
uv run ruff check .
# Run type checking
uv run mypy src/
# Run tests (when available)
uv run pytest
```
### Examples
```bash
# Basic workflow example
python example.py
# ReAct agent example
python examples/react_agent_example.py
# Planning agent example
python examples/planning_agent_example.py
# Sales chat agent
python examples/sales_chat_agent.py
```
## 📖 Documentation
For detailed documentation, see:
- **Architecture**: Understanding the framework design
- **Agent Types**: Comprehensive guide to different agent capabilities
- **Tool Development**: Creating custom tools and integrations
- **Memory Management**: Working with agent memory and persistence
- **LLM Integration**: Configuring and using different language models
## 🤝 Contributing
Contributions are welcome! Please:
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Commit your changes (`git commit -m 'Add amazing feature'`)
4. Push to the branch (`git push origin feature/amazing-feature`)
5. Open a Pull Request
### Development Guidelines
- Use type hints consistently
- Follow existing code patterns
- Add docstrings to public APIs
- Ensure thread safety for shared resources
- Test your changes thoroughly
## 📄 License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## 🙏 Acknowledgments
- OpenAI for GPT model integration
- Anthropic for Claude model support
- The Python async/await ecosystem
- Contributors and community feedback
---
**Aitor** - Build intelligent agents that think, plan, and act. 🤖✨
Raw data
{
"_id": null,
"home_page": null,
"name": "aitor",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "ai, workflow, builder, react, agents, llm",
"author": null,
"author_email": "Ashfakh <ashfakhrithu@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/0d/1e/486aa50e61dc9ebff3dae25a28cfdb117505cc9705261b6bb328b493442e/aitor-1.1.2.tar.gz",
"platform": null,
"description": "# Aitor\n\nAitor is a comprehensive Python framework for building intelligent, memory-enabled agents that can execute complex workflows with advanced reasoning capabilities. The framework combines DAG-based workflow execution with ReAct (Reasoning and Acting) agents, planning agents, and LLM integration to create powerful AI-driven applications.\n\n## \ud83d\ude80 Features\n\n### Core Framework\n- **Memory-Enabled Agents**: Stateful agents with typed memory management and persistence\n- **DAG Workflows**: Task dependency management through directed acyclic graphs\n- **Async Processing**: Thread-safe execution with both blocking (`ask`) and non-blocking (`tell`) APIs\n- **Task Chaining**: Intuitive `>>` operator for defining task dependencies\n\n### AI Agent Types\n- **ReAct Agents**: Reasoning and Acting agents with Think \u2192 Act \u2192 Observe loops\n- **Planning Agents**: Advanced agents that break down complex tasks into manageable todos\n- **Sub-Agent Management**: Delegate specialized tasks to focused sub-agents\n- **LLM Integration**: Support for OpenAI, Anthropic, and custom LLM providers\n\n### Advanced Capabilities\n- **Tool Registry**: Dynamic tool management with async execution\n- **Structured Responses**: JSON-based responses using Pydantic models\n- **Memory Persistence**: Export/import memory for session management\n- **Workflow Visualization**: Built-in Mermaid diagram generation\n\n## \ud83d\udce6 Installation\n\n```bash\npip install aitor\n```\n\nFor development:\n```bash\ngit clone https://github.com/Ashfakh/aitor.git\ncd aitor\npip install -e .\n```\n\n## \ud83c\udfc3 Quick Start\n\n### Basic Workflow Agent\n\n```python\nimport asyncio\nfrom typing import List\nfrom aitor import Aitor, Aitorflow, task\n\n@task\ndef clean_text(text: str) -> str:\n return text.strip().lower()\n\n@task\ndef count_words(text: str) -> int:\n return len(text.split())\n\n@task\ndef analyze_text(text: str, word_count: int) -> dict:\n return {\n \"text\": text,\n \"word_count\": word_count,\n \"avg_word_length\": len(text.replace(\" \", \"\")) / word_count if word_count > 0 else 0\n }\n\nasync def text_handler(message: str, aitor: Aitor[List[str]]):\n # Store in memory\n memory = aitor.get_memory()\n memory.append(message)\n aitor.set_memory(memory)\n \n # Execute workflow if attached\n if aitor.workflow:\n return await asyncio.to_thread(aitor.workflow.execute, message)\n return f\"Processed: {message}\"\n\nasync def main():\n # Create workflow\n workflow = Aitorflow(name=\"Text Analysis\")\n \n # Define task dependencies\n clean_text >> count_words >> analyze_text\n clean_text >> analyze_text # Multiple dependencies\n \n workflow.add_task(clean_text)\n \n # Create agent\n aitor = Aitor(\n initial_memory=[],\n name=\"TextProcessor\",\n on_receive_handler=text_handler\n )\n aitor.attach_workflow(workflow)\n \n # Process text\n result = await aitor.ask(\" Hello World Example! \")\n print(f\"Result: {result}\")\n \n Aitor.shutdown()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### ReAct Agent with Tools\n\n```python\nimport asyncio\nfrom aitor import create_react_agent\nfrom aitor.tools import tool\n\n@tool(name=\"calculator\", description=\"Perform mathematical calculations\")\ndef calculate(expression: str) -> float:\n \"\"\"Safely evaluate mathematical expressions.\"\"\"\n try:\n # Simple validation for safety\n allowed_chars = set('0123456789+-*/()., ')\n if not all(c in allowed_chars for c in expression):\n raise ValueError(\"Invalid characters in expression\")\n return eval(expression)\n except Exception as e:\n raise ValueError(f\"Calculation error: {e}\")\n\n@tool(name=\"text_analyzer\", description=\"Analyze text properties\")\ndef analyze_text(text: str) -> dict:\n \"\"\"Analyze various properties of text.\"\"\"\n return {\n \"length\": len(text),\n \"word_count\": len(text.split()),\n \"sentence_count\": text.count('.') + text.count('!') + text.count('?'),\n \"uppercase_ratio\": sum(1 for c in text if c.isupper()) / len(text) if text else 0\n }\n\nasync def main():\n # Create ReAct agent\n agent = await create_react_agent(\n name=\"MathTextAgent\",\n llm_provider=\"openai\",\n llm_config={\n \"api_key\": \"your-openai-api-key\",\n \"model\": \"gpt-4\"\n },\n tools=[calculate, analyze_text],\n max_reasoning_steps=10\n )\n \n # Solve complex problems\n response = await agent.solve(\n \"Calculate the square root of 144, then analyze the text 'Hello World!' \"\n \"and tell me the relationship between the calculation result and word count.\"\n )\n print(f\"Agent Response: {response}\")\n \n # Export memory for persistence\n memory_data = agent.export_memory()\n print(f\"Conversation history: {len(memory_data['conversation_history'])} messages\")\n \n await agent.shutdown()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### Planning Agent with Todos\n\n```python\nimport asyncio\nfrom aitor import PlanningReactAgent\nfrom aitor.llm import LLMManager\n\nasync def main():\n # Setup LLM\n llm_manager = LLMManager()\n llm_manager.add_provider(\n name=\"openai\",\n provider=\"openai\",\n config={\"api_key\": \"your-api-key\", \"model\": \"gpt-4\"}\n )\n \n # Create planning agent\n agent = PlanningReactAgent(\n name=\"ProjectPlanner\",\n llm_manager=llm_manager,\n max_reasoning_steps=20\n )\n \n # Complex planning task\n response = await agent.solve(\n \"Help me plan and execute a data analysis project. I need to collect data \"\n \"from APIs, clean it, perform statistical analysis, and create visualizations.\"\n )\n \n print(f\"Planning Response: {response}\")\n \n # Check created todos\n memory = agent.get_memory()\n print(f\"Created {len(memory.todos)} todos:\")\n for todo in memory.todos:\n print(f\" - [{todo.status}] {todo.title}\")\n \n await agent.shutdown()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## \ud83e\udde0 Agent Types\n\n### 1. Base Aitor Agent\nMemory-enabled agents with workflow integration:\n- Generic typed memory: `Aitor[T]`\n- Thread-safe operations\n- Workflow attachment\n- Async processing\n\n### 2. ReAct Agents\nReasoning and Acting agents that follow Think \u2192 Act \u2192 Observe loops:\n- **Tool Integration**: Dynamic tool registry with validation\n- **Reasoning Engine**: Step-by-step problem solving\n- **Memory Management**: Conversation and reasoning history\n- **LLM Integration**: Support for multiple providers\n\n### 3. Planning Agents\nAdvanced agents that break complex tasks into manageable todos:\n- **Todo Management**: Create, track, and execute todos with priorities\n- **Sub-Agent Delegation**: Spawn specialized agents for specific tasks\n- **Plan Execution**: Systematic approach to complex problems\n- **Progress Tracking**: Monitor todo completion and overall progress\n\n## \ud83d\udee0\ufe0f Core Components\n\n### Memory System\n```python\nfrom aitor.memory import ReactMemory\n\n# Structured memory with conversation, tools, and reasoning\nmemory = ReactMemory()\nmemory.add_message(\"user\", \"Hello!\")\nmemory.add_tool_execution(\"calculator\", {\"expression\": \"2+2\"}, 4)\nmemory.add_reasoning_step(\"THINK\", \"I need to solve this math problem\")\n```\n\n### Tool Registry\n```python\nfrom aitor.tools import ToolRegistry, tool\n\nregistry = ToolRegistry()\n\n@tool(name=\"example\", description=\"An example tool\")\ndef example_tool(param: str) -> str:\n return f\"Processed: {param}\"\n\nawait registry.register_tool(example_tool)\nresult = await registry.execute_tool(\"example\", {\"param\": \"test\"})\n```\n\n### LLM Management\n```python\nfrom aitor.llm import LLMManager\n\nllm_manager = LLMManager()\n\n# Add multiple providers\nllm_manager.add_provider(\"openai\", \"openai\", {\n \"api_key\": \"...\", \n \"model\": \"gpt-4\"\n})\nllm_manager.add_provider(\"claude\", \"anthropic\", {\n \"api_key\": \"...\", \n \"model\": \"claude-3-opus-20240229\"\n})\n\n# Switch between providers\nllm_manager.set_default_provider(\"claude\")\n```\n\n### Workflow Visualization\n```python\nfrom aitor import Aitorflow\n\nworkflow = Aitorflow(name=\"Example\")\n# ... add tasks ...\n\n# Generate Mermaid diagram\nmermaid_code = workflow.visualize()\nprint(mermaid_code)\n```\n\n## \ud83d\udcda Advanced Usage\n\n### Tool Creation Patterns\n\n#### 1. Basic Tool Creation\n```python\nfrom aitor.tools import tool\n\n@tool(\n name=\"calculator\",\n description=\"Perform mathematical calculations\",\n timeout=10.0,\n async_execution=True\n)\ndef calculate(expression: str) -> float:\n \"\"\"Calculate the result of a mathematical expression.\"\"\"\n return eval(expression)\n```\n\n#### 2. Tools with Pre-configured Parameters\nFor scenarios where you need to pass static parameters (like tenant details) that the agent doesn't provide:\n\n```python\nfrom aitor.tools import tool\n\ndef create_database_tool(tenant_id: str, connection_string: str):\n \"\"\"Factory function to create tenant-specific database tool.\"\"\"\n \n @tool(\n name=f\"database_query_{tenant_id}\",\n description=f\"Query database for tenant {tenant_id}\"\n )\n def query_database(query: str) -> dict:\n \"\"\"Execute database query with pre-configured tenant.\"\"\"\n # tenant_id and connection_string are captured from closure\n return execute_query(connection_string, tenant_id, query)\n \n return query_database\n\n# Usage during agent creation\ntenant_db_tool = create_database_tool(\"tenant_123\", \"postgresql://...\")\n```\n\n#### 3. Using functools.partial for Parameter Binding\n```python\nfrom functools import partial\nfrom aitor.tools import tool\n\ndef database_operation(tenant_id: str, api_key: str, query: str) -> dict:\n \"\"\"Base database function with all parameters.\"\"\"\n return {\n \"tenant\": tenant_id,\n \"query\": query,\n \"result\": f\"Query executed for {tenant_id}\"\n }\n\n# Create tenant-specific tool using partial\ntenant_specific_db = partial(\n database_operation, \n tenant_id=\"tenant_123\", \n api_key=\"secret_key\"\n)\n\n@tool(name=\"tenant_database\", description=\"Query database for current tenant\")\ndef tenant_db_tool(query: str) -> dict:\n return tenant_specific_db(query=query)\n```\n\n#### 4. Custom Tool Class with Context\n```python\nfrom aitor.tools import Tool\nfrom typing import Any, Dict\n\nclass TenantTool:\n \"\"\"Tool with pre-configured tenant parameters.\"\"\"\n \n def __init__(self, tenant_id: str, api_key: str, base_url: str):\n self.tenant_id = tenant_id\n self.api_key = api_key\n self.base_url = base_url\n \n def search_documents(self, query: str, limit: int = 10) -> dict:\n \"\"\"Search documents for this tenant.\"\"\"\n return {\n \"tenant\": self.tenant_id,\n \"query\": query,\n \"limit\": limit,\n \"results\": f\"Documents for {self.tenant_id}\"\n }\n \n def create_tool(self) -> Tool:\n \"\"\"Create the tool instance.\"\"\"\n return Tool(\n name=f\"search_docs_{self.tenant_id}\",\n func=self.search_documents,\n description=f\"Search documents for tenant {self.tenant_id}\",\n timeout=30.0\n )\n\n# Usage\ntenant_tool_factory = TenantTool(\"tenant_123\", \"api_key\", \"https://api.example.com\")\nsearch_tool = tenant_tool_factory.create_tool()\n```\n\n#### 5. Environment/Context-Based Configuration\n```python\nfrom aitor.tools import tool\nfrom dataclasses import dataclass\n\n@dataclass\nclass TenantContext:\n tenant_id: str\n database_url: str\n api_key: str\n storage_bucket: str\n\ndef create_tenant_tools(context: TenantContext):\n \"\"\"Create all tools for a specific tenant context.\"\"\"\n \n @tool(name=\"fetch_user_data\", description=\"Fetch user data for tenant\")\n def fetch_user_data(user_id: str) -> dict:\n # context is captured in closure\n return fetch_from_db(context.database_url, context.tenant_id, user_id)\n \n @tool(name=\"upload_file\", description=\"Upload file to tenant storage\")\n def upload_file(file_name: str, content: str) -> str:\n return upload_to_storage(context.storage_bucket, context.tenant_id, file_name, content)\n \n @tool(name=\"send_notification\", description=\"Send notification via tenant API\")\n def send_notification(message: str, recipient: str) -> bool:\n return send_via_api(context.api_key, context.tenant_id, message, recipient)\n \n return [fetch_user_data, upload_file, send_notification]\n\n# Usage\ntenant_context = TenantContext(\n tenant_id=\"tenant_123\",\n database_url=\"postgresql://...\",\n api_key=\"secret_123\",\n storage_bucket=\"tenant-123-files\"\n)\n\ntools = create_tenant_tools(tenant_context)\n\n# Register all tenant tools with agent\nfor tool in tools:\n await agent.register_tool(tool)\n```\n\n#### 6. Complete Tenant-Specific Agent Example\n```python\nfrom aitor import create_react_agent\nfrom aitor.tools import tool\n\nclass CustomerServiceAgent:\n def __init__(self, tenant_id: str, customer_db_url: str, support_api_key: str):\n self.tenant_id = tenant_id\n self.customer_db_url = customer_db_url\n self.support_api_key = support_api_key\n \n def create_tools(self):\n \"\"\"Create tenant-specific tools.\"\"\"\n \n @tool(name=\"lookup_customer\", description=\"Look up customer information\")\n def lookup_customer(customer_id: str) -> dict:\n return query_customer_db(self.customer_db_url, self.tenant_id, customer_id)\n \n @tool(name=\"create_ticket\", description=\"Create support ticket\")\n def create_ticket(title: str, description: str, priority: str = \"medium\") -> str:\n return create_support_ticket(\n self.support_api_key, \n self.tenant_id, \n title, \n description, \n priority\n )\n \n @tool(name=\"get_billing_info\", description=\"Get customer billing information\")\n def get_billing_info(customer_id: str) -> dict:\n return get_billing_data(self.customer_db_url, self.tenant_id, customer_id)\n \n return [lookup_customer, create_ticket, get_billing_info]\n \n async def create_agent(self, llm_manager):\n \"\"\"Create the complete agent with tenant-specific tools.\"\"\"\n tools = self.create_tools()\n \n agent = await create_react_agent(\n name=f\"CustomerServiceAgent_{self.tenant_id}\",\n llm_manager=llm_manager,\n tools=tools,\n agent_role=\"customer service representative\",\n additional_instructions=f\"You are helping customers for tenant {self.tenant_id}\"\n )\n \n return agent\n\n# Usage\ncs_agent_factory = CustomerServiceAgent(\n tenant_id=\"acme_corp\",\n customer_db_url=\"postgresql://...\",\n support_api_key=\"support_key_123\"\n)\n\nagent = await cs_agent_factory.create_agent(llm_manager)\n```\n\n#### Key Benefits of Pre-configured Tools:\n- **Security**: Sensitive credentials are encapsulated and not exposed to the agent\n- **Simplicity**: Agent only needs to pass business logic parameters\n- **Tenant Isolation**: Each agent instance is pre-configured for specific tenant\n- **Reusability**: Factory pattern allows creating multiple tenant-specific agents\n- **Type Safety**: Pre-configured parameters are validated at creation time\n\n### Avoiding Tool Registry Conflicts\n\nWhen creating multiple agents, ensure each has its own tool registry to prevent \"Tool already registered\" errors:\n\n```python\nfrom aitor import ReactAgentBuilder\nfrom aitor.tools import ToolRegistry\n\n# Method 1: Explicit tool registry per agent\ndef create_agent_with_tools(tenant_id: str):\n # Create fresh tool registry for this agent\n tool_registry = ToolRegistry()\n \n # Create tenant-specific tools\n @tool(name=\"lookup_info\", description=f\"Look up info for {tenant_id}\")\n def lookup_info(query: str) -> dict:\n return get_tenant_info(tenant_id, query)\n \n # Build agent with explicit registry\n agent = await (ReactAgentBuilder()\n .name(f\"Agent_{tenant_id}\")\n .tool_registry(tool_registry) # Explicit registry\n .add_tool(lookup_info)\n .build())\n \n return agent\n\n# Method 2: Using builder pattern (recommended)\nasync def create_isolated_agents():\n # Each agent gets its own tool registry automatically\n agent1 = await (ReactAgentBuilder()\n .name(\"Agent1\")\n .add_tool(create_lookup_tool(\"tenant1\"))\n .build())\n \n agent2 = await (ReactAgentBuilder()\n .name(\"Agent2\") \n .add_tool(create_lookup_tool(\"tenant2\")) # Same tool name, different registry\n .build())\n \n return agent1, agent2\n\n# Method 3: Factory pattern with unique tool names\ndef create_unique_tools(tenant_id: str):\n @tool(name=f\"lookup_info_{tenant_id}\", description=f\"Look up info for {tenant_id}\")\n def lookup_info(query: str) -> dict:\n return get_tenant_info(tenant_id, query)\n \n return lookup_info\n```\n\n### Memory Persistence\n```python\n# Export memory\nmemory_data = agent.export_memory()\n\n# Save to file\nimport json\nwith open(\"agent_memory.json\", \"w\") as f:\n json.dump(memory_data, f, indent=2)\n\n# Load and import\nwith open(\"agent_memory.json\", \"r\") as f:\n memory_data = json.load(f)\n\nnew_agent.import_memory(memory_data)\n```\n\n### Sub-Agent Architecture\n```python\nfrom aitor.sub_agent import SubAgentManager\n\n# Create specialized sub-agents\nsub_manager = SubAgentManager()\nsub_manager.create_sub_agent(\n name=\"DataAnalyst\",\n specialization=\"statistical analysis and data processing\",\n tools=[pandas_tool, numpy_tool, matplotlib_tool]\n)\n\n# Delegate tasks\nresult = await sub_manager.delegate_task(\n \"DataAnalyst\",\n \"Analyze this dataset and create visualizations\"\n)\n```\n\n## \ud83c\udfaf Use Cases\n\n- **Data Processing Pipelines**: Complex ETL workflows with error handling\n- **AI-Powered Assistants**: Conversational agents with tool access\n- **Automated Planning**: Break down complex projects into actionable steps\n- **Research Automation**: Gather, analyze, and synthesize information\n- **Code Generation**: AI agents that write and test code\n- **Content Creation**: Multi-step content workflows with reviews\n\n## \ud83e\uddea Development\n\n### Setup Development Environment\n```bash\n# Clone repository\ngit clone https://github.com/Ashfakh/aitor.git\ncd aitor\n\n# Create virtual environment\npython -m venv venv\nsource venv/bin/activate # Windows: venv\\Scripts\\activate\n\n# Install in development mode\npip install -e .\n\n# Install development dependencies\npip install -e .[dev]\n```\n\n### Running Tests\n```bash\n# Run linting\nuv run ruff check .\n\n# Run type checking\nuv run mypy src/\n\n# Run tests (when available)\nuv run pytest\n```\n\n### Examples\n```bash\n# Basic workflow example\npython example.py\n\n# ReAct agent example\npython examples/react_agent_example.py\n\n# Planning agent example\npython examples/planning_agent_example.py\n\n# Sales chat agent\npython examples/sales_chat_agent.py\n```\n\n## \ud83d\udcd6 Documentation\n\nFor detailed documentation, see:\n- **Architecture**: Understanding the framework design\n- **Agent Types**: Comprehensive guide to different agent capabilities\n- **Tool Development**: Creating custom tools and integrations\n- **Memory Management**: Working with agent memory and persistence\n- **LLM Integration**: Configuring and using different language models\n\n## \ud83e\udd1d Contributing\n\nContributions are welcome! Please:\n\n1. Fork the repository\n2. Create a feature branch (`git checkout -b feature/amazing-feature`)\n3. Commit your changes (`git commit -m 'Add amazing feature'`)\n4. Push to the branch (`git push origin feature/amazing-feature`)\n5. Open a Pull Request\n\n### Development Guidelines\n- Use type hints consistently\n- Follow existing code patterns\n- Add docstrings to public APIs\n- Ensure thread safety for shared resources\n- Test your changes thoroughly\n\n## \ud83d\udcc4 License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## \ud83d\ude4f Acknowledgments\n\n- OpenAI for GPT model integration\n- Anthropic for Claude model support\n- The Python async/await ecosystem\n- Contributors and community feedback\n\n---\n\n**Aitor** - Build intelligent agents that think, plan, and act. \ud83e\udd16\u2728\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python framework for building intelligent, memory-enabled agents with ReAct reasoning",
"version": "1.1.2",
"project_urls": {
"Bug Tracker": "https://github.com/Ashfakh/aitor/issues",
"Homepage": "https://github.com/Ashfakh/aitor"
},
"split_keywords": [
"ai",
" workflow",
" builder",
" react",
" agents",
" llm"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "af7e72815d2851e6c4ae4f76411979d00b8c00a84d49aa1352a13561ca386e78",
"md5": "c7ca0b1861e80c9cf234126a55d1123e",
"sha256": "7cd3b358733817ab7643e4f45b0602e648ca6d7b732f34de054f94b5330bf1fc"
},
"downloads": -1,
"filename": "aitor-1.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c7ca0b1861e80c9cf234126a55d1123e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 54193,
"upload_time": "2025-07-23T10:50:16",
"upload_time_iso_8601": "2025-07-23T10:50:16.280401Z",
"url": "https://files.pythonhosted.org/packages/af/7e/72815d2851e6c4ae4f76411979d00b8c00a84d49aa1352a13561ca386e78/aitor-1.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0d1e486aa50e61dc9ebff3dae25a28cfdb117505cc9705261b6bb328b493442e",
"md5": "5b40ea263b2accef2951f454a85d6298",
"sha256": "aad5a9d2ff13b51ae74dbd4ca5ac926bd92f91150db3134b03b384e1aa8c9112"
},
"downloads": -1,
"filename": "aitor-1.1.2.tar.gz",
"has_sig": false,
"md5_digest": "5b40ea263b2accef2951f454a85d6298",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 53931,
"upload_time": "2025-07-23T10:50:17",
"upload_time_iso_8601": "2025-07-23T10:50:17.556990Z",
"url": "https://files.pythonhosted.org/packages/0d/1e/486aa50e61dc9ebff3dae25a28cfdb117505cc9705261b6bb328b493442e/aitor-1.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-23 10:50:17",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Ashfakh",
"github_project": "aitor",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aitor"
}
