Aura Local Agent

Local AI orchestration framework combining Ollama, state management, and deterministic task execution.

The Problem

Building autonomous AI systems requires:

• Multiple Components: LLMs, state management, task queues, external APIs
• Integration Friction: Custom glue code to connect them
• Dependency Complexity: Managing authentication, configurations, failures
• Debugging Difficulty: Hard to trace execution flow across systems

Traditional approaches require cloud APIs, constant internet connection, and complex deployment pipelines.

The Solution

Aura is a modular, local-first orchestration framework that:

✅ Runs Entirely Local - Ollama LLM + Firebase mock + local storage
✅ Zero API Dependencies - No external services required
✅ Event-Driven - Clean event listeners for system integration
✅ Modular Components - Mix and match based on your needs
✅ Production-Ready - Structured logging, error handling, state persistence

Architecture

graph TB
    Input["Input<br/>(Any Source)"] -->|Listen| Agent["Aura Agent<br/>(Orchestrator)"]
    Agent -->|Process| LLM["Ollama LLM<br/>(Local)"]
    Agent -->|State| State["Firebase Mock<br/>(Local JSON)"]
    Agent -->|Execute| Executor["Task Executor<br/>(Automations)"]
    
    Executor -->|Emit| Tasks["Task Results"]
    Tasks -->|Store| State
    
    style Agent fill:#4a90e2
    style LLM fill:#50e3c2
    style State fill:#f5a623
    style Executor fill:#b8e986

Loading

Quick Start

Prerequisites

• Python 3.8+
• Ollama running locally on http://localhost:11434
• Mistral model: ollama pull mistral

Installation

git clone https://github.com/buubear14/aura-local-agent.git
cd aura-local-agent
pip install -r requirements.txt

First Example

python examples/example_basic_usage.py

Expected output:

📊 Agent Status:
{
  "state": "idle",
  "pending_tasks": 0,
  "completed_tasks": 0,
  "llm_available": true
}

🔄 Processing cycle:
📥 Input received: {"type": "lead_submission", ...}
🧠 LLM processed: 45 tokens

📋 Task queued: task_1716...
⚙️  Executing tasks:
✓ Task Completed: task_1716... (log_lead)

Core Components

1. Config Manager (src/config.py)

Central configuration for all components.

from src.config import CONFIG

print(CONFIG.ollama.model)  # "mistral"
print(CONFIG.firebase.use_mock)  # True

2. Ollama Wrapper (src/ollama_wrapper.py)

Interface to local Ollama with structured prompting.

from src.ollama_wrapper import OllamaWrapper

ollama = OllamaWrapper()
response = ollama.structured_extraction(
    data="Company: TechCorp, Employees: 150",
    schema={"company": "string", "employees": "number"}
)
print(response.output)  # Valid JSON

3. Firebase Mock (src/firebase_state.py)

Local state persistence without Firebase credentials.

from src.firebase_state import StateManager

state = StateManager()
state.save_task("task_1", {"company": "TechCorp", "tier": "Hot"})
task = state.get_task("task_1")  # Retrieve anytime

4. Task Executor (src/task_executor.py)

Handles automation task execution.

from src.task_executor import TaskExecutor, AutomationTask

executor = TaskExecutor()
task = AutomationTask("task_1", "send_email", {"to": "user@example.com"})
result = executor.execute(task)
print(result.status)  # TaskStatus.COMPLETED

5. Aura Agent (src/agent.py)

Main orchestration engine combining all components.

from src.agent import get_agent

agent = get_agent()

# Listen for input
agent.listen_for_input(lambda: {"type": "lead_received", ...})

# Execute pending tasks
executed = agent.execute_pending_tasks()

# Check status
print(agent.get_status())

Event-Driven Architecture

Register event listeners for custom logic:

from src.agent import get_agent

agent = get_agent()

# Listen for task received
agent.on("task_received", lambda task: print(f"New task: {task.task_id}"))

# Listen for task completed
agent.on("task_completed", lambda task: print(f"Completed: {task.result}"))

# Listen for context updates
agent.on("context_updated", lambda ctx: print(f"Context: {ctx}"))

Examples

Example 1: Basic Usage

python examples/example_basic_usage.py

Shows complete flow: input → LLM processing → task execution

Example 2: Firebase Integration

python examples/example_firebase_integration.py

Demonstrates state persistence and querying

Supported Task Types

The TaskExecutor supports these automation types out of the box:

Task Type	Purpose	Mock Behavior
send_email	Send emails	Logs to console
log_lead	Record lead data	Stores in state
update_crm	Update CRM record	Logs update
schedule_demo	Schedule meeting	Generates meet link
webhook_call	Call HTTP webhook	Logs to console
database_write	Write to database	Records in state

Register custom handlers:

def my_custom_task(params):
    return {"status": "completed", "result": params}

agent.task_executor.register_handler("my_task", my_custom_task)

State Persistence

All state is automatically saved to .aura/firebase_state.json:

{
  "tasks": {
    "task_001": {
      "data": {"company": "TechCorp", "tier": "Hot"},
      "created_at": "2024-01-15T10:30:00",
      "updated_at": "2024-01-15T10:35:00"
    }
  },
  "agent": {
    "context": {
      "last_analysis": "...",
      "confidence": 85
    }
  }
}

Configuration

Configure via environment variables or config.py:

export OLLAMA_BASE_URL="http://localhost:11434"
export OLLAMA_MODEL="mistral"
export OLLAMA_TEMPERATURE="0"
export FIREBASE_USE_MOCK="true"
export AURA_DEBUG="true"
export AURA_LOG_LEVEL="DEBUG"

Or programmatically:

from src.config import ConfigManager

config = ConfigManager()
config.ollama.model = "neural-chat"
config.agent.debug = True

Integration Patterns

Pattern 1: Webhook Handler

from src.agent import get_agent
from flask import Flask

app = Flask(__name__)
agent = get_agent()

@app.route('/webhook', methods=['POST'])
def webhook():
    agent.listen_for_input(lambda: request.json)
    agent.execute_pending_tasks()
    return {"status": "processed"}

Pattern 2: Scheduled Jobs

from src.agent import get_agent
from apscheduler.schedulers.background import BackgroundScheduler

agent = get_agent()
scheduler = BackgroundScheduler()

def check_for_tasks():
    agent.execute_pending_tasks()

scheduler.add_job(check_for_tasks, 'interval', minutes=5)
scheduler.start()

Pattern 3: CLI Interface

from src.agent import get_agent
import click

agent = get_agent()

@click.command()
@click.argument('input_file')
def process(input_file):
    with open(input_file) as f:
        data = json.load(f)
    
    agent.listen_for_input(lambda: data)
    agent.execute_pending_tasks()
    print(agent.get_status())

if __name__ == '__main__':
    process()

Limitations & Considerations

1. Local Only: Requires Ollama running locally
2. Single Machine: State is local JSON, not distributed
3. Model Dependent: Quality depends on Mistral's capabilities
4. Context Window: Limited to Mistral's 8K token context

Next Steps

1. Try Example 1 to see basic flow
2. Modify task handlers for your use cases
3. Register custom event listeners
4. Integrate with your data sources
5. Deploy to production with proper monitoring

Tech Stack

• LLM: Ollama (Mistral)
• State: Local JSON files
• Language: Python 3.8+
• Async: Standard library (threading ready)
• Logging: Python logging module

Contributing

Contributions welcome! Areas for enhancement:

• Additional task executors (SMS, Slack, Teams)
• Async/parallel task execution
• Advanced state querying
• Performance optimizations
• Cloud backend support (optional)

License

MIT License - See LICENSE file

Author

Adriaan du Randt - Prompt Engineering & Automation Specialist

• GitHub: @buubear14
• Email: adriaandurandt@gmail.com

Related Projects

• Structural Prompting Framework: Deterministic prompt engineering
• Information Broker: B2B lead generation pipeline
• Agent-47: Modular Gemini CLI automation

---

Start with Example 1 to see Aura in action.