-
Notifications
You must be signed in to change notification settings - Fork 2
Autonomous Agent Guide
Creator: Mahesh Vaijainthymala Krishnamoorthy (Mahesh Vaikri)
This guide covers MAPLE's autonomous agentic AI capabilities introduced in v1.1.0.
MAPLE's autonomy layer adds LLM-powered reasoning on top of the production infrastructure. The AutonomousAgent extends the base Agent with:
- ReAct Loop — Think, Act, Reflect cycle for goal pursuit
- Tool Use — LLM selects and executes tools to accomplish tasks
- Memory — Three-tier memory system for context, history, and knowledge
- Reflection — Periodic self-assessment with backtracking when stuck
- Human-in-the-Loop — Approval callbacks for sensitive tool execution
AutonomousAgent
┌──────────────────────────────┐
│ pursue_goal("description") │
│ │ │
│ ┌─────▼─────┐ │
│ │ THINK │ LLM reasons │
│ └─────┬─────┘ │
│ │ │
│ ┌─────▼─────┐ │
│ │ ACT │ Execute tools │
│ └─────┬─────┘ │
│ │ │
│ ┌─────▼─────┐ │
│ │ REFLECT │ Assess/back │
│ └─────┬─────┘ track │
│ │ │
│ Done? ──► Return Goal │
└──────────────────────────────┘
from maple import AutonomousAgent, AutonomousConfig, LLMConfig
config = AutonomousConfig(
llm=LLMConfig(
provider="openai", # "openai" or "anthropic"
model="gpt-4",
api_key="sk-...", # or set OPENAI_API_KEY env var
temperature=0.7,
max_tokens=4096
),
max_reasoning_steps=20, # Max think-act-reflect cycles
working_memory_tokens=8000, # Context window budget
reflection_frequency=5 # Reflect every N steps
)
agent = AutonomousAgent("my_agent", autonomy_config=config)result = agent.pursue_goal("Analyze the codebase and identify performance bottlenecks")
if result.is_ok():
goal = result.unwrap()
print(f"Status: {goal.status}") # "completed"
print(f"Result: {goal.result}")
print(f"Steps: {len(goal.reasoning_trace)}")
else:
print(f"Failed: {result.unwrap_err()}")from maple import Tool, Result
def search_web(query: str) -> Result:
# Your search implementation
results = my_search_api(query)
return Result.ok(results)
def write_file(path: str, content: str) -> Result:
with open(path, 'w') as f:
f.write(content)
return Result.ok(f"Written to {path}")
# Register tools
agent.register_tool(Tool(
name="search_web",
description="Search the web for information",
parameters={
"query": {"type": "string", "description": "Search query"}
},
handler=search_web
))
agent.register_tool(Tool(
name="write_file",
description="Write content to a file",
parameters={
"path": {"type": "string"},
"content": {"type": "string"}
},
handler=write_file,
requires_approval=True # Human must approve
))AutonomousAgent automatically registers these tools from MAPLE infrastructure:
| Tool | Description |
|---|---|
send_message |
Send a message to another agent |
query_agents |
Find agents by capability via AgentRegistry |
read_state |
Read from distributed StateStore |
write_state |
Write to distributed StateStore |
check_resources |
Query available resources via ResourceManager |
establish_link |
Create a secure communication link |
def approval_handler(tool_name: str, arguments: dict) -> bool:
print(f"Agent wants to use '{tool_name}' with args: {arguments}")
response = input("Approve? (y/n): ")
return response.lower() == 'y'
agent.set_approval_callback(approval_handler)Tools with requires_approval=True will call this callback before execution. You can also set require_approval_for=["write_file", "send_email"] in AutonomousConfig.
from maple import MemoryManager
# Access via agent
memory = agent.memory_manager
# Working Memory — current context window
memory.working.add("search_results", "Found 5 relevant papers", relevance=0.9)
context = memory.working.get_context()
# Episodic Memory — past task interactions
memory.episodic.record("task_123", {"action": "searched", "result": "5 papers"})
history = memory.episodic.recall("task_123")
# Semantic Memory — learned facts
memory.semantic.store_fact("project_language", "Python 3.12")
lang = memory.semantic.recall_fact("project_language")The ReAct loop automatically:
- Adds tool results to working memory
- Records each step in episodic memory
- When working memory is full, summarizes and archives to episodic memory using the LLM
One agent decomposes the goal and delegates sub-goals to workers.
from maple.autonomy.orchestrator import AgentOrchestrator
orchestrator = AgentOrchestrator()
# Form a team
team_id = orchestrator.form_team(
"research_team",
agents=[researcher, analyst, writer],
roles=["worker", "worker", "supervisor"]
).unwrap()
# Supervisor decomposes goal, assigns to workers, collects results
result = orchestrator.execute_supervised(team_id, "Write a market analysis report")All agents independently work on the same question, then a supervisor synthesizes.
result = orchestrator.execute_consensus(team_id, "What are the top 3 risks for Q2?")team_id = orchestrator.form_team_by_capability(
"auto_team",
required_capabilities=["research", "writing", "analysis"],
available_agents=[agent1, agent2, agent3, agent4]
).unwrap()Every step of the ReAct loop is logged.
from maple.autonomy.observability import DecisionLogger, AgentSnapshot
# Get all decision traces for a goal
logger = agent.decision_logger
traces = logger.get_traces(goal_id="goal_123")
for trace in traces:
print(f"Step {trace.step_number}: {trace.response_summary}")
print(f" Tools: {trace.tool_calls}")
print(f" Tokens: {trace.token_usage}")
print(f" Duration: {trace.duration_ms}ms")
# Export as JSON
json_data = logger.export_json(goal_id="goal_123")snapshot = AgentSnapshot.capture(agent)
# Returns: metrics, working memory, active goals, LLM usage statsconfig = LLMConfig(
provider="openai",
model="gpt-4",
api_key="sk-..."
)
# For local models (vLLM, Ollama, etc.)
config = LLMConfig(
provider="openai",
model="my-local-model",
api_base="http://localhost:8000/v1",
api_key="not-needed"
)config = LLMConfig(
provider="anthropic",
model="claude-sonnet-4-20250514",
api_key="sk-ant-..."
)from maple.llm.provider import LLMProvider
from maple.llm.registry import LLMProviderRegistry
class MyProvider(LLMProvider):
def complete(self, messages, tools=None, **kwargs):
# Your implementation
pass
LLMProviderRegistry.register("my_provider", MyProvider)Discover and use tools from external MCP servers.
from maple.autonomy.mcp_tools import discover_mcp_tools, register_mcp_tools
# Discover tools from an MCP server
tools = discover_mcp_tools("http://localhost:3000", agent).unwrap()
# Register them in the agent's tool registry
count = register_mcp_tools(agent.tool_registry, tools)
print(f"Registered {count} MCP tools")See the examples directory:
-
hello_autonomous_agent.py— Basic autonomous agent with custom tools -
multi_agent_team.py— Multi-agent orchestration with supervisor pattern
- Home — Project overview
- Installation Guide — Setup and configuration
- API Reference — Complete API documentation
- Protocol Specification v1.0 — Core protocol specification
Created by: Mahesh Vaijainthymala Krishnamoorthy (Mahesh Vaikri)
MAPLE v1.1.0 | Copyright (C) 2025-2026 Mahesh Vaijainthymala Krishnamoorthy (Mahesh Vaikri) | License AGPL 3.0
MAPLE v1.1.0
Links