A sophisticated AI Agent framework that combines LLM capabilities with tool-based execution for complex task automation.
- 🛠️ Tool System: Web scraping, file processing, text manipulation, calculations
- 🧠 Smart Planning: Automatic task analysis and workflow creation
- 💾 Memory System: Caching and recall of similar tasks
- 🔄 Adaptive Processing: Chooses between direct LLM vs tool-based execution
- ⚡ Error Handling: Robust error recovery and logging
- Python 3.8+
- OpenRouter API account
- Clone and install dependencies:
pip install -r requirments.txt- Environment Configuration:
Create a
.envfile:
api-key=your_openrouter_api_key_here
LLM-model=your_preferred_modelfrom enhanced_agent import run_enhanced_agent
import asyncio
async def main():
tasks = [
"Reverse this text: hello world",
"Calculate 45 * 12 + 8",
"Download content from https://example.com",
]
results = await run_enhanced_agent(tasks)
print(results)
asyncio.run(main())from enhanced_agent import EnhancedAIAgent
async def custom_agent():
agent = EnhancedAIAgent(api_key="your_key", model="your_model")
# Process complex tasks
result = await agent.process_task(
"Find all laptop prices from https://divar-contest.darkube.app/divar_sample.html and sum them"
)
print(result)| Tool | Description | Usage Example |
|---|---|---|
web_downloader |
Download web content | "Get content from https://example.com" |
file_downloader |
Download files | "Download file from URL" |
text_processor |
Text manipulation | "Reverse this sentence" |
calculator |
Mathematical operations | "Calculate 125 * 8" |
url_extractor |
Extract URLs from text | Auto-used when URLs detected |
User Task
↓
Task Analyzer (AgentBrain)
↓
Workflow Planner (TaskPlanner)
↓
Tool Executor (TaskExecutor)
↓
Result Formatter
↓
Final Output
- AgentBrain: Analyzes tasks and determines required tools
- TaskPlanner: Creates execution workflows
- TaskExecutor: Orchestrates tool execution
- ToolRegistry: Manages available tools
- AgentMemory: Caches results and learns patterns
task = "Reverse this text: Hello AI World"
result = await agent.process_task(task)
# Returns: "World AI Hello"task = "Extract headings from https://example.com"
result = await agent.process_task(task)
# Returns list of headingstask = "What is 3*5 + 2//3?"
result = await agent.process_task(task)
# Returns: "15"task = "Download https://divar-contest.darkube.app/data.txt and sum all numbers"
# Agent automatically: Extracts URL → Downloads → Parses → Calculates → Returns sumctf_tasks = [
"do what image says at { https://divar-contest.darkube.app/fyvkr93-public.png }",
"find the first laptop in https://divar-contest.darkube.app/divar_sample.html",
"calculate 3*5 + 2 // 3",
".tfel nruter tsuj t'nod tub rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw"
]
results = await run_enhanced_agent(ctf_tasks)api-key: Your OpenRouter API keyLLM-model: Model name (e.g., "anthropic/claude-3-sonnet")
- Temperature: 0.1 (for consistent outputs)
- Max Tokens: 512
- Timeout: 30 seconds
The system includes comprehensive error handling:
- Tool execution failures
- Network timeouts
- Invalid responses
- Memory cache misses
- Requires OpenRouter API access
- Limited to available tools
- Async operations required
- Web scraping depends on target site permissions
def my_custom_tool(**kwargs):
# Tool implementation
return result
# Register tool
agent.tools.register_tool(
name="my_tool",
function=my_custom_tool,
description="What my tool does"
)For issues and questions:
- Check your API key and model settings
- Verify internet connectivity for web tools
- Ensure all dependencies are installed
