4.5 Prompt Patterns for Tool Invocation

# Recommended project structure for agent development agent-system/ ├── docs/ # Documentation and specifications │ ├── architecture.md │ ├── requirements.md │ └── api-reference.md ├── src/ # Source code │ ├── agents/ # Individual agent implementations │ │ ├── coordinator.py │ │ ├── research_agent.py │ │ └── analysis_agent.py │ ├── common/ # Shared utilities and base classes │ │ ├── base_agent.py │ │ ├── message_bus.py │ │ └── config.py │ ├── tools/ # External tool integrations │ │ ├── web_search.py │ │ ├── database.py │ │ └── api_clients.py │ └── interfaces/ # User interfaces and APIs │ ├── rest_api.py │ ├── web_ui.py │ └── cli.py ├── tests/ # Test suites │ ├── unit/ # Unit tests │ ├── integration/ # Integration tests │ ├── performance/ # Performance tests │ └── e2e/ # End-to-end tests ├── deployment/ # Deployment configurations │ ├── docker-compose.yml │ ├── kubernetes/ │ └── terraform/ ├── monitoring/ # Monitoring and observability │ ├── prometheus.yml │ ├── grafana-dashboards/ │ └── alerts.yml ├── requirements.txt # Python dependencies ├── Dockerfile # Container definition ├── README.md # Project overview and setup └── .github/workflows/ # CI/CD pipeline definitions

# Base agent class with development best practices from abc import ABC, abstractmethod import logging import uuid from typing import Dict, Any, Optional from dataclasses import dataclass import asyncio @dataclass class AgentMessage: id: str sender: str recipient: str content: Dict[str, Any] message_type: str timestamp: float class BaseAgent(ABC): """Base class for all agents with common functionality""" def __init__( self, agent_id: str, config: Dict[str, Any] ): self.agent_id = agent_id self.config = config self.logger = logging.getLogger( f"agent.{agent_id}" ) self.state = "initialized" self.metrics = { "messages_processed": 0, "errors": 0, "start_time": None } async def start(self): """Start the agent""" self.state = "running" self.metrics["start_time"] = asyncio.get_event_loop().time() self.logger.info( f"Agent {self.agent_id} started" ) await self.on_start() async def stop(self): """Stop the agent gracefully""" self.state = "stopping" await self.on_stop() self.state = "stopped" self.logger.info( f"Agent {self.agent_id} stopped" ) async def handle_message( self, message: AgentMessage ) -> Optional[AgentMessage]: """Handle incoming message with error handling and metrics""" try: self.logger.debug( f"Processing message {message.id}" ) result = await self.process_message(message) self.metrics["messages_processed"] += 1 return result except Exception as e: self.metrics["errors"] += 1 self.logger.error( f"Error processing message: {e}" ) await self.on_error(e, message) return None # Abstract methods that subclasses must implement @abstractmethod async def process_message( self, message: AgentMessage ) -> Optional[AgentMessage]: """Process a message and return response if needed""" pass # Hook methods for customization async def on_start(self): """Called when agent starts""" pass async def on_stop(self): """Called when agent stops""" pass async def on_error( self, error: Exception, message: AgentMessage ): """Called when error occurs during message processing""" pass def get_metrics(self) -> Dict[str, Any]: """Get agent performance metrics""" return { "agent_id": self.agent_id, "state": self.state, **self.metrics }