File size: 8,811 Bytes
dc893fb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 | """Example 4: Full Agent with All Features
This example demonstrates a complete agent setup with:
- All basic tools (Read, Write, Edit, Bash)
- Session Note tools for memory
- MCP tools (Memory, Search, etc.)
- Skills integration
Based on: tests/test_integration.py
"""
import asyncio
import tempfile
from pathlib import Path
from mini_agent import LLMClient
from mini_agent.agent import Agent
from mini_agent.config import Config
from mini_agent.tools import BashTool, EditTool, ReadTool, WriteTool
from mini_agent.tools.mcp_loader import load_mcp_tools_async
from mini_agent.tools.note_tool import RecallNoteTool, SessionNoteTool
async def demo_full_agent():
"""Demo: Full-featured agent with all capabilities."""
print("\n" + "=" * 60)
print("Full Mini Agent - All Features Enabled")
print("=" * 60)
# Load configuration
config_path = Path("mini_agent/config/config.yaml")
if not config_path.exists():
print("❌ config.yaml not found. Please run:")
print(" cp mini_agent/config/config-example.yaml mini_agent/config/config.yaml")
return
config = Config.from_yaml(config_path)
# Check API key
if not config.llm.api_key or config.llm.api_key.startswith("YOUR_"):
print("❌ API key not configured in config.yaml")
return
# Create workspace
with tempfile.TemporaryDirectory() as workspace_dir:
print(f"📁 Workspace: {workspace_dir}")
# Load system prompt (Agent will auto-inject workspace info)
system_prompt_path = Path("mini_agent/config/system_prompt.md")
if system_prompt_path.exists():
system_prompt = system_prompt_path.read_text(encoding="utf-8")
else:
system_prompt = "You are a helpful AI assistant."
# Add Session Note instructions
note_instructions = """
IMPORTANT - Session Memory:
You have record_note and recall_notes tools. Use them to:
- Save important facts, decisions, and context
- Recall previous information across conversations
"""
system_prompt += note_instructions
# Initialize LLM
llm_client = LLMClient(
api_key=config.llm.api_key,
api_base=config.llm.api_base,
model=config.llm.model,
)
# Initialize basic tools
tools = [
ReadTool(workspace_dir=workspace_dir),
WriteTool(workspace_dir=workspace_dir),
EditTool(workspace_dir=workspace_dir),
BashTool(),
]
print("✓ Loaded 4 basic tools")
# Add Session Note tools
memory_file = Path(workspace_dir) / ".agent_memory.json"
tools.extend(
[
SessionNoteTool(memory_file=str(memory_file)),
RecallNoteTool(memory_file=str(memory_file)),
]
)
print("✓ Loaded 2 Session Note tools")
# Load MCP tools (if configured)
try:
mcp_tools = await load_mcp_tools_async(config_path="mini_agent/config/mcp.json")
if mcp_tools:
tools.extend(mcp_tools)
print(f"✓ Loaded {len(mcp_tools)} MCP tools")
else:
print("⚠️ No MCP tools configured (mcp.json is empty or disabled)")
except Exception as e:
print(f"⚠️ MCP tools not loaded: {e}")
# Create agent
agent = Agent(
llm_client=llm_client,
system_prompt=system_prompt,
tools=tools,
max_steps=config.agent.max_steps,
workspace_dir=workspace_dir,
)
print(f"\n🤖 Agent created with {len(tools)} total tools\n")
# Task: Complex task that uses multiple tools
task = """
Please help me with the following tasks:
1. Create a Python script called 'calculator.py' that:
- Has functions for add, subtract, multiply, divide
- Has a main() function that demonstrates usage
- Includes proper docstrings and type hints
2. Create a README.md file that:
- Describes the calculator script
- Shows how to run it
- Lists the available functions
3. Test the calculator by running it with bash
4. Remember this project info:
- Project: Simple Calculator
- Language: Python
- Purpose: Demonstration of agent capabilities
"""
print("=" * 60)
print("📝 Task:")
print("=" * 60)
print(task)
print("\n" + "=" * 60)
print("🤖 Agent is working...\n")
agent.add_user_message(task)
try:
result = await agent.run()
print("\n" + "=" * 60)
print("✅ Agent completed!")
print("=" * 60)
print(f"\nAgent's final response:\n{result}\n")
# Show created files
print("=" * 60)
print("📄 Created files in workspace:")
print("=" * 60)
workspace = Path(workspace_dir)
for file in workspace.glob("*"):
if file.is_file() and not file.name.startswith("."):
print(f"\n📄 {file.name}:")
print("-" * 60)
content = file.read_text()
# Show first 20 lines
lines = content.split("\n")[:20]
print("\n".join(lines))
if len(content.split("\n")) > 20:
print("... (truncated)")
print("-" * 60)
# Show memory
if memory_file.exists():
import json
notes = json.loads(memory_file.read_text())
print(f"\n💾 Session notes recorded: {len(notes)}")
for note in notes:
print(f" - [{note['category']}] {note['content'][:60]}...")
except Exception as e:
print(f"❌ Error during agent execution: {e}")
import traceback
traceback.print_exc()
async def demo_interactive_mode():
"""Demo: Interactive conversation with agent."""
print("\n" + "=" * 60)
print("Interactive Mini Agent")
print("=" * 60)
print("\nThis demo shows multi-turn conversation.")
print("(In production, use `mini-agent` for full interactive mode)")
# Load config
config_path = Path("mini_agent/config/config.yaml")
if not config_path.exists():
print("❌ config.yaml not found")
return
config = Config.from_yaml(config_path)
if not config.llm.api_key or config.llm.api_key.startswith("YOUR_"):
print("❌ API key not configured")
return
with tempfile.TemporaryDirectory() as workspace_dir:
# Setup
system_prompt = "You are a helpful assistant with access to tools."
llm_client = LLMClient(
api_key=config.llm.api_key,
api_base=config.llm.api_base,
model=config.llm.model,
)
tools = [
WriteTool(workspace_dir=workspace_dir),
ReadTool(workspace_dir=workspace_dir),
BashTool(),
]
agent = Agent(
llm_client=llm_client,
system_prompt=system_prompt,
tools=tools,
max_steps=20,
workspace_dir=workspace_dir,
)
# Conversation turns
conversations = [
"Create a file called 'data.txt' with the numbers 1 to 5, one per line.",
"Now read the file and tell me what's in it.",
"Count how many lines are in the file using bash.",
]
for i, message in enumerate(conversations, 1):
print(f"\n{'=' * 60}")
print(f"Turn {i}:")
print(f"{'=' * 60}")
print(f"User: {message}\n")
agent.add_user_message(message)
try:
result = await agent.run()
print(f"Agent: {result}\n")
except Exception as e:
print(f"Error: {e}")
break
async def main():
"""Run all demos."""
print("=" * 60)
print("Full Agent Examples")
print("=" * 60)
print("\nThese examples demonstrate the complete agent capabilities:")
print("- All basic tools (file operations, bash)")
print("- Session memory (persistent notes)")
print("- MCP tools (if configured)")
print("- Multi-turn conversations\n")
# Run demos
await demo_full_agent()
print("\n" * 2)
await demo_interactive_mode()
print("\n" + "=" * 60)
print("All demos completed! ✅")
print("=" * 60)
print("\n💡 Next step: Try the interactive mode with:")
print(" mini-agent\n")
if __name__ == "__main__":
asyncio.run(main())
|