Ever had that moment when your production system goes haywire, and you're frantically searching through documentation trying to remember if this was the incident that needed a database rollback or a cache flush? Yeah, me too. Let's talk about how we can use Language Learning Models (LLMs) to make our lives easier when dealing with incidents.
The Multi-Agent Vision
What's better than one AI assistant? Multiple AI assistants! Today we're diving into a fascinating approach: using multiple specialised agents to handle different types of incidents. Think of it as having a team of super-specialised first responders, each one an expert in their domain. It's like having a SWAT team, firefighters, and paramedics all ready to jump in — but in this case, they're AI agents.
Why Multiple Agents?
You might be thinking, "Why not just dump all the runbooks into one super-agent and call it a day?" (I know I was tempted!). But there's a method to our madness, and it comes down to three key benefits:
- Specialised knowledge: Each agent has a focused system prompt containing detailed instructions for handling specific types of incidents. This specialisation allows for more accurate and relevant responses compared to a generalist approach. With growing context windows, you could argue this is important, but if your runbooks are really big and detailed, then you could be hitting the limit sooner than you wish!
- Tool access control: Different types of incidents will require different tools. Our security incident agent might need access to log analysis and network isolation controls, while our service outage agent needs service restart capabilities and load balancer controls. Giving every agent access to every tool is like giving every employee a master key to the building — probably not the best idea! To me, this is one of the key benefits.
- Clear responsibility boundaries: The coordinator handles routing while specialised agents handle specific incident types. This separation of concerns makes the system easier to maintain and update.
The Architecture
At the heart of our system is what a "Traffic Controller" — a coordinator agent that knows exactly which specialist to call for each type of incident. When an alert comes in, this coordinator doesn't panic (unlike some of us humans may do). Instead, it calmly analyses the situation and routes it to the right specialist.
The Conversation Flow
One of the trickier aspects to figure out was the conversation flow. Should every follow-up question bounce through the coordinator? We solved this with a "full handoff" approach:
- The coordinator receives the initial incident report
- It determines the type of incident and selects the appropriate specialist
- The specialist takes over the conversation, maintaining its own context and state
- The specialist has access only to its designated tools
This approach reduces latency (no need to route every message through the coordinator) and allows for better context maintenance.
Implementation Details
Our implementation uses OpenAI's GPT-4 model, but you could adapt it for any LLM. We've organised the code into a clean, modular structure that separates concerns and makes it easy to add new specialist types or tools.
The system consists of:
- A base agent class that handles LLM interactions
- A coordinator agent that routes incidents
- Specialist agents with their own system prompts and tool access
- A tool registry that defines which tools are available to each specialist
- Models for incidents and conversations
Configuration and Setup
All you need to get started is Python 3.8+ and an OpenAI API key. Put your API key in a .env file:
OPENAI_API_KEY=your-key-here
Adding New Specialists
Want to add a new type of specialist? Just:
- Add the new incident type to the
IncidentTypeenum - Create a system prompt for your specialist
- Define which tools they can access
- Register the specialist with the coordinator
Results and Considerations
I'm still working on implementing this system, but I am hoping to gain the following benefits:
- More precise responses due to specialisation
- Better tool access control
- Easier to audit and maintain
- Specialists can be optimised independently
The Actual Code
Project Structure
First, let's look at how we'll organise our code:
incident_response/
│
├── .env # Environment variables (API keys)
├── requirements.txt # Project dependencies
├── README.md # Project documentation
│
├── src/
│ ├── __init__.py
│ ├── agents/
│ │ ├── __init__.py
│ │ ├── base.py # Base agent class
│ │ ├── coordinator.py # Coordinator agent
│ │ └── specialist.py # Specialist agents
│ │
│ ├── models/
│ │ ├── __init__.py
│ │ ├── incident.py # Incident data models
│ │ └── conversation.py # Conversation handling
│ │
│ └── tools/
│ ├── __init__.py
│ └── incident_tools.py # Tool implementations
│
└── examples/
└── handle_incident.py # Usage examples
src/models/incident.py
from enum import Enum
from typing import List
from pydantic import BaseModel
from datetime import datetime
class IncidentType(Enum):
SECURITY_BREACH = "security_breach" # When someone tries to be sneaky
SERVICE_OUTAGE = "service_outage" # The classic "why is everything down?!"
DATA_CORRUPTION = "data_corruption" # The "oh no, the data!" moment
class Incident(BaseModel):
"""Represents an incident - because every crisis needs structure!"""
type: IncidentType
description: str
severity: int # On a scale of "meh" to "DEFCON 1"
affected_systems: List[str]
created_at: datetime = datetime.now()
class Message(BaseModel):
"""A message in the conversation"""
content: str
timestamp: datetime = datetime.now()
from_user: bool
class ConversationState(str, Enum):
INITIAL = "initial"
SPECIALIST_ENGAGED = "specialist_engaged"
RESOLVED = "resolved"
class Conversation(BaseModel):
"""Tracks the state of an incident response conversation"""
id: str
messages: List[Message] = []
state: ConversationState = ConversationState.INITIAL
current_agent: str = None
src/tools/incident_tools.py
class IncidentTools:
"""Available tools for incident response"""
@staticmethod
def check_logs(system: str, timeframe: str) -> str:
"""Simulated log checking functionality"""
return f"Logs retrieved for {system} during {timeframe}"
@staticmethod
def restart_service(service_name: str) -> str:
"""Simulated service restart"""
return f"Service {service_name} restarted"
@staticmethod
def query_metrics(metric: str, duration: str) -> str:
"""Simulated metrics query"""
return f"Metrics for {metric} over {duration}"
@staticmethod
def isolate_system(system: str) -> str:
"""Simulated system isolation"""
return f"System {system} isolated from network"
@staticmethod
def validate_backup(backup_id: str) -> str:
"""Simulated backup validation"""
return f"Backup {backup_id} validated successfully"
class ToolRegistry:
"""Registry of available tools for each incident type"""
SECURITY_TOOLS = ["check_logs", "isolate_system", "query_metrics"]
OUTAGE_TOOLS = ["restart_service", "query_metrics", "check_logs"]
DATA_TOOLS = ["validate_backup", "query_metrics", "check_logs"]
src/agents/base.py
import os
from typing import List, Dict
from openai import OpenAI
from dotenv import load_dotenv
from ..models.incident import Conversation, Message
from ..tools.incident_tools import IncidentTools
# Load environment variables
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class BaseAgent:
def __init__(self, system_prompt: str, allowed_tools: List[str]):
self.system_prompt = system_prompt
self.allowed_tools = allowed_tools
self.tools = IncidentTools()
def _build_messages(self, conversation: Conversation) -> List[Dict]:
"""Build the messages array for the OpenAI API"""
messages = [
{"role": "system", "content": self.system_prompt}
]
# Add available tools context
tools_context = f"Available tools: {', '.join(self.allowed_tools)}"
messages.append({
"role": "system",
"content": tools_context
})
# Add conversation history
for msg in conversation.messages[-5:]: # Last 5 messages for context
role = "user" if msg.from_user else "assistant"
messages.append({"role": role, "content": msg.content})
return messages
def generate_response(self, conversation: Conversation) -> str:
"""Generate a response using the OpenAI API"""
messages = self._build_messages(conversation)
formatted_messages = []
for msg in messages:
formatted_messages.append({
"role": msg["role"],
"content": [{"type": "text", "text": msg["content"]}]
})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=formatted_messages,
temperature=0.7,
max_tokens=500
)
return response.choices[0].message.content
src/agents/specialist.py
from typing import Dict
from .base import BaseAgent
from ..models.incident import Conversation, Message, ConversationState
class SpecialistAgent(BaseAgent):
def __init__(self, specialty: str, system_prompt: str, allowed_tools: List[str]):
super().__init__(system_prompt, allowed_tools)
self.specialty = specialty
def handle_message(self, message: str, conversation: Conversation) -> str:
"""Handle a message in the context of the ongoing conversation"""
# Add the new message to conversation
conversation.messages.append(Message(
content=message,
from_user=True
))
# Generate response
response = self.generate_response(conversation)
# Add response to conversation history
conversation.messages.append(Message(
content=response,
from_user=False
))
return response
# Pre-configured specialist prompts
SPECIALIST_PROMPTS = {
"security": """You are a security incident response specialist.
Think of yourself as the digital equivalent of a master detective.
Follow these steps for security breaches:
1. Isolate affected systems (quarantine the suspicious ones!)
2. Assess breach scope (how bad is it, really?)
3. Contain the breach (stop the bleeding)
4. Collect forensic data (gather evidence)
5. Remediate vulnerabilities (close those doors!)""",
"reliability": """You are a service reliability expert.
The person everyone calls when things go dark.
Follow these steps for service outages:
1. Check monitoring dashboards (where's the fire?)
2. Identify failure points (find the culprit)
3. Execute recovery procedures (work your magic)
4. Verify service restoration (is it really fixed?)
5. Document root cause (what did we learn?)""",
"data": """You are a data recovery specialist.
The last hope for corrupted data everywhere.
Follow these steps for data corruption:
1. Stop affected processes (freeze everything!)
2. Assess corruption scope (how much data is having a bad day?)
3. Restore from backups (you did backup, right?)
4. Verify data integrity (trust but verify)
5. Implement prevention measures (never again!)"""
}
src/agents/coordinator.py
from typing import Dict
from .base import BaseAgent
from .specialist import SpecialistAgent, SPECIALIST_PROMPTS
from ..models.incident import Conversation, Message, ConversationState, IncidentType
from ..tools.incident_tools import ToolRegistry
class CoordinatorAgent:
"""The traffic controller of our incident response system"""
def __init__(self):
self.conversations: Dict[str, Conversation] = {}
# Initialize specialist agents
self.specialists = {
IncidentType.SECURITY_BREACH: SpecialistAgent(
"security",
SPECIALIST_PROMPTS["security"],
ToolRegistry.SECURITY_TOOLS
),
IncidentType.SERVICE_OUTAGE: SpecialistAgent(
"reliability",
SPECIALIST_PROMPTS["reliability"],
ToolRegistry.OUTAGE_TOOLS
),
IncidentType.DATA_CORRUPTION: SpecialistAgent(
"data",
SPECIALIST_PROMPTS["data"],
ToolRegistry.DATA_TOOLS
)
}
# Initialize coordinator's own LLM capabilities
self.agent = BaseAgent(
system_prompt="""You are a coordinator for incident response.
Your job is to analyse incoming incident reports and determine
which specialist should handle them. Focus on identifying the
type of incident and its severity.""",
allowed_tools=[]
)
def determine_incident_type(self, message: str) -> IncidentType:
"""Use the coordinator's LLM to determine incident type"""
system_prompt = """You are an incident coordinator. Your job is to determine the type of incident based on the user's description. The possible incident types are:
1. SECURITY_BREACH - Any security-related incidents including unauthorized access, suspicious activities, potential breaches, unusual login patterns, etc.
2. SERVICE_OUTAGE - Any availability issues including system downtime, performance degradation, service unavailability, connection problems, etc.
3. DATA_CORRUPTION - Any data integrity issues including corrupt files, database inconsistencies, missing data, data quality issues, etc.
Respond ONLY with the incident type that best matches the description. Only respond with one of: SECURITY_BREACH, SERVICE_OUTAGE, or DATA_CORRUPTION"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Determine the incident type for this situation: {message}"}
]
formatted_messages = [
{
"role": "system",
"content": [{"type": "text", "text": system_prompt}]
},
{
"role": "user",
"content": [{
"type": "text",
"text": f"Determine the incident type for this situation: {message}"
}]
}
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=formatted_messages,
temperature=0, # Use 0 for more deterministic responses
max_tokens=50 # We only need a short response
)
incident_type_str = response.choices[0].message.content.strip()
# Map the response to our IncidentType enum
try:
return IncidentType(incident_type_str.lower())
except ValueError:
# Default to data corruption if we get an unexpected response
# In production, you might want to handle this differently
print(f"Warning: Unexpected incident type '{incident_type_str}', defaulting to data corruption")
return IncidentType.DATA_CORRUPTION
def handle_message(self, conversation_id: str, message: str) -> str:
"""Handle an incoming message"""
# Get or create conversation
if conversation_id not in self.conversations:
self.conversations[conversation_id] = Conversation(
id=conversation_id
)
conversation = self.conversations[conversation_id]
if conversation.state == ConversationState.INITIAL:
# Determine which specialist should handle this
incident_type = self.determine_incident_type(message)
specialist = self.specialists[incident_type]
# Update conversation state
conversation.current_agent = specialist.specialty
conversation.state = ConversationState.SPECIALIST_ENGAGED
return specialist.handle_message(message, conversation)
elif conversation.state == ConversationState.SPECIALIST_ENGAGED:
# Continue conversation with current specialist
specialist = self.specialists[
IncidentType(conversation.current_agent)
]
return specialist.handle_message(message, conversation)
examples/handle_incident.py
from src.agents.coordinator import CoordinatorAgent
def main():
# Initialize the coordinator
coordinator = CoordinatorAgent()
# Simulate a security incident conversation
conv_id = "incident-123"
messages = [
"We've detected unusual access patterns in our authentication logs",
"Yes, there are multiple failed login attempts from the same IP",
"Should we block these IPs and force password resets?"
]
# Process each message
for message in messages:
print(f"\nUser: {message}")
response = coordinator.handle_message(conv_id, message)
print(f"Assistant: {response}")
if __name__ == "__main__":
main()
Call the example by running:
python examples/handle_incident.py
Final Thoughts
Whether this approach is right for you depends on your specific needs. Consider:
- How complex are your incidents?
- How critical is response time?
- What's your tolerance for additional system complexity?
- How important is tool access control in your environment?
Remember, there's no one-size-fits-all solution. The best approach is often the one that matches your organisation's specific needs and constraints. Just make sure you're not creating more complexity than you're solving!
Ready to try it yourself? The complete code is above. Feel free to adapt it to your needs, and don't forget to let me know how it works for you!
Disclaimer: The perspectives shared here are my own and do not necessarily represent those of my employer. I use GenAI as a tool to help me compose and structure my articles.