hermes-agent-architecture

# Hermes Agent Architecture > Skill by [ara.so](https://ara.so) — Hermes Skills collection. Hermes Agent is a production-grade LLM agent framework by Nous Research featuring advanced memory management, multi-agent orchestration, 18+ messaging platform integrations, and a sophisticated tool execution system. This skill covers internal architecture, extension patterns, and implementation strategies verified against source code. ## Installation ```bash # Clone the repository git clone https://github.com/NousResearch/hermes-agent.git cd hermes-agent # Install dependencies pip install -e . # Or with Poetry poetry install # Basic configuration cp config.example.yaml config.yaml # Edit config.yaml with your API keys and preferences ``` ## Core Architecture Components ### Agent Loop and Execution The main agent loop is in `hermes/agent.py`: ```python from hermes.agent import Agent from hermes.config import Config # Initialize agent config = Config.load("config.yaml") agent = Agent(config) # Run interactive session await agent.run() # Programmatic execution response = await agent.process_message( "Analyze the repository structure", context={"cwd": "/path/to/repo"} ) ``` **Key execution flow:** 1. `process_message()` → Prompt assembly 2. Model inference → Tool calls extraction 3. Tool dispatch via `ToolRegistry` 4. Result aggregation → Memory storage 5. Response generation ### Tool System Architecture Tools are registered centrally via decorators: ```python from hermes.tools.registry import tool_registry from hermes.tools.base import ToolResult @tool_registry.register( name="custom_analyzer", description="Analyze code patterns", category="analysis", parameters={ "file_path": { "type": "string", "description": "Path to file to analyze" }, "pattern": { "type": "string", "description": "Pattern to search for" } } ) async def custom_analyzer(file_path: str, pattern: str, **kwargs) -> ToolResult: """Custom code analysis tool.""" try: with open(file_path, 'r') as f: content = f.read() matches = re.findall(pattern, content) return ToolResult( success=True, data={"matches": matches, "count": len(matches)}, message=f"Found {len(matches)} matches" ) except Exception as e: return ToolResult( success=False, error=str(e) ) ``` **Toolset grouping** (from `hermes/tools/toolsets.py`): ```python from hermes.tools.toolsets import Toolset, toolset_registry @toolset_registry.register("code_analysis") class CodeAnalysisToolset(Toolset): """Custom toolset for code analysis.""" def get_tools(self): return [ "custom_analyzer", "list_functions", "complexity_check" ] def get_description(self): return "Tools for analyzing code structure and patterns" ``` ### Memory System Three-layer architecture (`hermes/memory/`): ```python from hermes.memory.manager import MemoryManager from hermes.memory.store import MemoryStore from hermes.memory.provider import MemoryProvider # Initialize memory system store = MemoryStore(db_path="~/.hermes/memory.db") manager = MemoryManager(store) # Store interaction await manager.add_message( role="user", content="Remember that I prefer functional programming", session_id="current_session" ) # Retrieve relevant memories memories = await manager.search_memories( query="programming preferences", limit=5 ) # Freeze snapshot for prompt caching snapshot = manager.freeze_snapshot() # This protects the prefix cache boundary ``` **Session search with FTS5:** ```python from hermes.memory.session_db import SessionDB session_db = SessionDB(db_path="~/.hermes/sessions.db") # Search across sessions results = await session_db.search( query="docker configuration", limit=10 ) # Get LLM summary of related sessions summary = await session_db.get_session_summary( query="docker issues", llm_client=auxiliary_client ) ``` ### Context Compression v3 Automatic context management (`hermes/compression/compressor.py`): ```python from hermes.compression.compressor import ContextCompressor compressor = ContextCompressor( model_client=client, max_tokens=128000, preserve_recent=5 # Keep last 5 messages uncompressed ) # Three-stage preprocessing compressed = await compressor.compress( messages=conversation_history, strategies=[ "md5_dedup", # Remove duplicate tool results "smart_collapse", # Collapse similar adjacent messages "param_truncation" # Truncate large parameters ] ) # Structured summarization summary = await compressor.summarize_structured( messages=old_messages, format="bullet_points" # or "narrative" ) ``` ### Skills System Progressive disclosure with conditional activation (`hermes/skills/`): ```python from hermes.skills.manager import SkillsManager skills_manager = SkillsManager( skills_dir="~/.hermes/skills", config=config ) # Skills are auto-discovered from markdown files # Triggered by keywords or explicit @skill references # Conditional activation example in YAML frontmatter: """ --- name: docker-expert triggers: - docker - container - dockerfile conditions: - file_exists: Dockerfile - OR: - file_exists: docker-compose.yml - env_var: DOCKER_HOST credentials: - DOCKER_API_KEY --- """ # Plugin namespace skills (loaded from plugins) await skills_manager.load_plugin_skills( plugin_name="custom_plugin", skills_manifest=plugin.get_skills() ) ``` ### Multi-Agent Architecture Four runtime mechanisms: ```python # 1. Task Delegation from hermes.tools.delegate import delegate_task result = await delegate_task( task="Research Python async patterns", specialist_config={ "model": "claude-3-7-sonnet", "toolsets": ["web_search", "code_analysis"] } ) # 2. Mixture of Agents (MoA) from hermes.multi_agent.moa import MixtureOfAgents moa = MixtureOfAgents( agents=[ {"name": "researcher", "model": "gpt-4"}, {"name": "critic", "model": "claude-3-opus"}, {"name": "synthesizer", "model": "claude-3-7-sonnet"} ] ) consensus = await moa.deliberate( question="What's the best architecture for this service?" ) # 3. Background Review from hermes.multi_agent.reviewer import BackgroundReviewer reviewer = BackgroundReviewer(model="gpt-4o") review = await reviewer.review_conversation( messages=conversation_history, focus="security concerns" ) # 4. Direct Agent Messaging await agent.send_message( to_agent="code_reviewer", content="Please review the changes in PR #123" ) ``` ### Browser Automation Multi-backend architecture (`hermes/tools/browser/`): ```python from hermes.tools.browser import browser_navigate, browser_interact # Navigate with accessibility tree extraction result = await browser_navigate( url="https://github.com/trending", extract_content=True, backend="playwright" # or "selenium", "playwright_firefox" ) # Interact with elements await browser_interact( action="click", selector="button[aria-label='Star']", wait_for="networkidle" ) # Three-layer security: # 1. URL allowlist/blocklist # 2. Content filtering # 3. Sandboxed execution ``` ### Code Execution Sandbox Secure Python execution (`hermes/tools/code_exec/`): ```python from hermes.tools.code_exec import execute_code result = await execute_code( code=""" import numpy as np data = np.random.rand(100) print(f"Mean: {data.mean()}") """, language="python", timeout=30, allowed_imports=["numpy", "pandas", "matplotlib"] ) # Sandbox restrictions: # - No os.system, subprocess, eval # - Limited file system access # - Network requests blocked by default # - Resource limits enforced ``` **Communication modes:** ```python # 1. Unix Domain Socket (default) sandbox_config = { "mode": "uds", "socket_path": "/tmp/hermes_sandbox.sock" } # 2. File RPC (Windows-compatible) sandbox_config = { "mode": "file_rpc", "rpc_dir": "/tmp/hermes_rpc" } ``` ### Messaging Gateway Integration Platform adapter plugin system (`hermes/gateway/`): ```python from hermes.gateway.platform_registry import platform_registry from hermes.gateway.base import PlatformAdapter, PlatformMessage @platform_registry.register("custom_chat") class CustomChatAdapter(PlatformAdapter): """Custom messaging platform integration.""" platform_name = "custom_chat" async def initialize(self): """Connect to platform API.""" self.client = CustomChatClient( api_key=self.config.get("api_key") ) await self.client.connect() async def receive_messages(self): """Poll for new messages.""" async for raw_msg in self.client.stream_messages(): yield PlatformMessage( platform="custom_chat", channel_id=raw_msg.channel, user_id=raw_msg.author_id, username=raw_msg.author_name, content=raw_msg.text, message_id=raw_msg.id, timestamp=raw_msg.created_at ) async def send_message(self, channel_id: str, content: str, **kwargs): """Send response to platform.""" await self.client.send( channel=channel_id, text=content ) def get_channel_prompt(self, channel_id: str) -> str: """Optional: platform-specific instructions.""" return "Respond in a friendly, casual tone suitable for chat." # Register and run gateway = MessagingGateway(config) gateway.register_platform(CustomChatAdapter(config.platforms.custom_chat)) await gateway.run() ``` **Built-in platform adapters:** - Discord, Slack, Telegram, IRC - WeChat, QQ, DingTalk, WeCom (企业微信) - WhatsApp, Signal, Matrix - BlueBubbles (iMessage), SMS - 腾讯元宝 (Tencent Yuanbao) ### Plugin System Dual hook architecture (`hermes/plugins/`): ```python from hermes.plugins.base import Plugin, plugin_registry @plugin_registry.register class DashboardPlugin(Plugin): """Web dashboard for monitoring agent activity.""" name = "dashboard" version = "1.0.0" async def initialize(self, agent): """Setup plugin.""" self.agent = agent self.app = create_dashboard_app() # Register custom commands agent.register_command( name="/dashboard", handler=self.open_dashboard, description="Open web dashboard" ) # Hook into tool execution agent.register_hook( "before_tool_call", self.log_tool_call ) async def log_tool_call(self, tool_name, parameters): """Log tool executions to dashboard.""" await self.app.broadcast_event({ "type": "tool_call", "tool": tool_name, "params": parameters, "timestamp": time.time() }) async def open_dashboard(self, args): """Handle /dashboard command.""" url = await self.app.get_url() return f"Dashboard: {url}" # Load plugins await agent.load_plugins(plugins_dir="~/.hermes/plugins") ``` ### MCP (Model Context Protocol) Integration ```python from hermes.mcp.client import MCPClient # Connect to MCP server mcp = MCPClient(server_url="http://localhost:8000") # MCP tools automatically registered await mcp.connect() mcp_tools = await mcp.list_tools() # Tools appear in agent's tool registry # OAuth flows handled automatically for supported MCPs ``` ### Smart Model Routing ```python from hermes.routing.smart_router import SmartRouter router = SmartRouter( default_model="claude-3-7-sonnet", short_message_model="claude-3-5-haiku", short_message_threshold=100 # tokens ) # Automatic routing based on complexity model = router.select_model( messages=conversation, task_type="code_generation" # or "chat", "analysis" ) # Provider-specific features # - AWS Bedrock with cross-region failover # - Gemini with OAuth refresh # - Ollama Cloud distributed routing # - Tool Gateway for model-specific tool schemas ``` ### Prompt Caching Optimization