Notes pratiques : Créez votre premier agent IA DevOps à l’aide de LangChain et d’outils Bash
Guide pas à pas pratique : Créez votre premier agent AI DevOps en utilisant LangChain et des outils Bash : contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes qui mettent en œuvre ce modèle.
Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : Créer votre premier agent AI DevOps à l’aide de LangChain et d’outils Bash. L’accent est mis sur des étapes opérationnelles, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner son intention. L’auteur doit définir l’étape d’aperçu, les entrées nécessaires, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Documentez conjointement le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’une mise en forme ultérieure.
Qu’est-ce que nous construisons vraiment ?
Lors de l’étape « Que construisons-nous ? », notez d’abord les exigences : les entrées nécessaires, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité et non un processus embrouillé. Enregistrez le nom de l’outil, le hash des arguments, la latence et le résultat de chaque appel. Sans ces traces, le débogage de boucles d’agent prend des heures inutilement.
Chatbot: "Your nginx config might have a syntax error."
Agent: [runs `nginx -t`] → "Confirmed. Line 42 in /etc/nginx/sites-enabled/app.conf
has an unclosed bracket. Here's the fix."
Pourquoi LangChain pour cela ?
Lorsque vous travaillez sur l’étape « Pourquoi LangChain ? », notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès, et refusez les terminations partielles silencieuses. Enregistrez le nom de l’outil, le hash des arguments, la latence et le résultat de chaque appel. Sans ce suivi, le débogage des boucles d’agent prend des heures inutilement.
Prérequis
Lors de la phase des Prérequis, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Journalisez le nom outil, l’hash des arguments, la latence et le résultat de chaque appel. Sans ce suivi, les boucles d’agent de débogage gaspillent des heures.
# Python 3.11+
python --version
# Install dependencies
pip install \\
langchain \\
langchain-anthropic \\
langchain-community \\
anthropic \\
python-dotenv \\
--break-system-packages
# Set your API key
export ANTHROPIC_API_KEY="your-key-here"
# Or create a .env file
echo "ANTHROPIC_API_KEY=your-key-here" > .env
Lors de la phase des Prérequis, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Documentez en même temps le parcours normal et le parcours de récupération. Les tentatives de réessai, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations ultérieures.
devops-agent/
├── agent/
│ ├── __init__.py
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── bash_tools.py # Core bash execution tools
│ │ ├── kubernetes_tools.py # kubectl wrapper tools
│ │ ├── log_tools.py # Log reading and analysis
│ │ └── system_tools.py # System info tools
│ ├── agent.py # Agent definition and loop
│ └── safety.py # Command safety checks
├── config.py
├── main.py
└── .env
Boucle de l’agent ReAct
La phase du cycle d’agent ReAct fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé. Exposez des outils dotés de schémas restreints et de labels explicites indiquant les effets secondaires. Les hôtes doivent savoir quels appels modifient l’état avant d’approuver automatiquement.
User: "Why is the API service slow?"
Agent Loop:
┌─────────────────────────────────────────────┐
│ THOUGHT: I need to check the service first │
│ ACTION: run_command("systemctl status api")│
│ OBSERVE: service is running, high CPU │
├─────────────────────────────────────────────┤
│ THOUGHT: High CPU - check what's running │
│ ACTION: run_command("top -bn1 | head -20")│
│ OBSERVE: api process at 95% CPU │
├─────────────────────────────────────────────┤
│ THOUGHT: Check recent logs for errors │
│ ACTION: read_logs("/var/log/api/app.log") │
│ OBSERVE: Massive DB query loop in logs │
├─────────────────────────────────────────────┤
│ THOUGHT: I have enough info now │
│ FINAL ANSWER: "The API is slow because..." │
└─────────────────────────────────────────────┘
Étape 1 — La sécurité d’abord
La phase « Sécurité d’abord » du Étape 1 fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette phase comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès et refusez toute mise en œuvre partielle silencieuse. Exposez des outils dotés de schémas restreints et de labels explicites indiquant leurs effets secondaires. Les hôtes doivent savoir quels appels modifient l’état avant d’approuver automatiquement.
import re
from typing import Tuple
# Commands that are NEVER allowed regardless of context
BLOCKED_COMMANDS = [
r"rm\\s+-rf\\s+/", # rm -rf /
r"dd\\s+if=", # disk wipe
r"mkfs\\.", # format filesystem
r">\\s*/dev/sd", # write to disk
r"chmod\\s+-R\\s+777\\s+/", # open all permissions
r"passwd\\s+root", # change root password
r"userdel\\s+", # delete users
r"iptables\\s+-F", # flush all firewall rules
r"shutdown", # shutdown/reboot
r"reboot",
r"halt",
r"curl.*\\|\\s*bash", # pipe curl to bash
r"wget.*\\|\\s*bash", # pipe wget to bash
r"eval\\s+", # eval injection
r"base64\\s+--decode.*\\|", # decode and execute
]
# Commands that require extra caution (logged but allowed)
SENSITIVE_COMMANDS = [
"sudo", "su ", "ssh ", "scp ", "rsync",
"systemctl stop", "systemctl disable",
"apt remove", "yum remove", "pip uninstall",
"kubectl delete", "kubectl drain",
"terraform destroy", "ansible-playbook"
]
# Read-only commands - always safe
READONLY_COMMANDS = [
"cat ", "less ", "head ", "tail ", "grep ",
"find ", "ls ", "ps ", "top ", "df ", "du ",
"netstat", "ss ", "lsof ", "curl -s",
"systemctl status", "systemctl list",
"kubectl get", "kubectl describe", "kubectl logs",
"docker ps", "docker inspect", "docker logs",
"ansible --list-hosts",
"terraform plan", "terraform show",
"aws ec2 describe", "aws s3 ls",
"git log", "git status", "git diff",
"ping ", "traceroute", "nslookup", "dig "
]
def check_command_safety(command: str) -> Tuple[bool, str, str]:
"""
Returns: (is_allowed, risk_level, reason)
risk_level: "safe" | "sensitive" | "blocked"
"""
command_lower = command.lower().strip()
# Check blocked patterns
for pattern in BLOCKED_COMMANDS:
if re.search(pattern, command_lower):
return False, "blocked", f"Command matches blocked pattern: {pattern}"
# Check sensitive commands
for sensitive in SENSITIVE_COMMANDS:
if sensitive in command_lower:
return True, "sensitive", f"Sensitive command detected: {sensitive}"
# Check if it's read-only
for readonly in READONLY_COMMANDS:
if command_lower.startswith(readonly):
return True, "safe", "Read-only command"
# Default: allow but flag as unknown
return True, "unknown", "Command not in any list - proceeding with caution"
Étape 2 — Outils Bash de base
La phase Step 2 Core Bash fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et une note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de la démonstration aux environnements partagés. Exposez des outils dotés de schémas restreints et de labels explicites indiquant les effets secondaires. Les hôtes doivent savoir quels appels modifient l’état avant d’approuver automatiquement. La phase Step 2 Core Bash fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et une note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours réussi et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure.
import subprocess
from langchain_core.tools import tool
from agent.safety import check_command_safety
import logging
logger = logging.getLogger(__name__)
@tool
def run_command(command: str) -> str:
"""
Run a bash command and return its output.
Use this to check system status, read configs, inspect processes.
Examples:
- run_command("systemctl status nginx")
- run_command("ps aux | grep python")
- run_command("df -h")
- run_command("netstat -tlnp")
"""
is_allowed, risk_level, reason = check_command_safety(command)
if not is_allowed:
return f"BLOCKED: Command not allowed. Reason: {reason}"
if risk_level == "sensitive":
logger.warning(f"SENSITIVE command executed: {command}")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
output = result.stdout or result.stderr
# Truncate very long output
if len(output) > 3000:
output = output[:3000] + "\\n... [output truncated]"
if result.returncode != 0:
return f"Command failed (exit {result.returncode}):\\n{output}"
return output or "(no output)"
except subprocess.TimeoutExpired:
return "Command timed out after 30 seconds"
except Exception as e:
return f"Error running command: {str(e)}"
@tool
def read_file(file_path: str) -> str:
"""
Read the contents of a file.
Use this to inspect configs, logs, scripts.
Examples:
- read_file("/etc/nginx/nginx.conf")
- read_file("/var/log/syslog")
- read_file("/etc/systemd/system/myapp.service")
"""
try:
with open(file_path, 'r', errors='replace') as f:
content = f.read()
# Truncate large files
if len(content) > 4000:
# Show beginning and end for log files
content = content[:2000] + "\\n\\n... [middle truncated] ...\\n\\n" + content[-2000:]
return content
except PermissionError:
return f"Permission denied: {file_path}"
except FileNotFoundError:
return f"File not found: {file_path}"
except Exception as e:
return f"Error reading file: {str(e)}"
@tool
def read_logs(service_name: str, lines: int = 50) -> str:
"""
Read recent logs for a systemd service using journalctl.
Use this to diagnose service errors, crashes, or warnings.
Examples:
- read_logs("nginx")
- read_logs("docker", lines=100)
- read_logs("postgresql")
"""
command = f"journalctl -u {service_name} -n {lines} --no-pager -o short-iso"
try:
result = subprocess.run(
command, shell=True,
capture_output=True, text=True, timeout=15
)
return result.stdout or result.stderr or "(no logs found)"
except Exception as e:
return f"Error reading logs: {str(e)}"
@tool
def grep_logs(log_path: str, pattern: str, lines_before: int = 2, lines_after: int = 2) -> str:
"""
Search for a pattern in a log file with context lines.
Use this to find specific errors, IPs, or events in logs.
Examples:
- grep_logs("/var/log/nginx/error.log", "upstream timed out")
- grep_logs("/var/log/auth.log", "Failed password", lines_after=0)
"""
command = f"grep -i -B {lines_before} -A {lines_after} '{pattern}' {log_path} | tail -100"
is_allowed, _, _ = check_command_safety(f"grep {log_path}")
if not is_allowed:
return "BLOCKED: Cannot read this log file"
try:
result = subprocess.run(
command, shell=True,
capture_output=True, text=True, timeout=20
)
return result.stdout or f"No matches found for '{pattern}' in {log_path}"
except Exception as e:
return f"Error searching logs: {str(e)}"
Étape 3 — Outils Kubernetes
L’auteur de l’étape 3 des outils Kubernetes définit les entrées, le responsable de l’étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un pipeline embrouillé. Authentifiez-vous au niveau du gateway et réautorisez-vous au niveau du data plane. Un token porteur seul ne constitue pas une frontière entre les tenants.
import subprocess
from langchain_core.tools import tool
def kubectl(command: str) -> str:
"""Run a kubectl command safely."""
# Only allow read operations from the agent
read_verbs = ["get", "describe", "logs", "top", "explain", "version", "cluster-info"]
cmd_parts = command.strip().split()
if cmd_parts and cmd_parts[0] not in read_verbs:
return f"BLOCKED: Only read operations allowed. Got: {cmd_parts[0]}"
result = subprocess.run(
f"kubectl {command}",
shell=True, capture_output=True, text=True, timeout=30
)
return result.stdout or result.stderr
@tool
def k8s_get_pods(namespace: str = "default") -> str:
"""
List all pods in a namespace with their status.
Use this to check if pods are running, crashing, or pending.
Examples:
- k8s_get_pods("production")
- k8s_get_pods("monitoring")
"""
return kubectl(f"get pods -n {namespace} -o wide")
@tool
def k8s_describe_pod(pod_name: str, namespace: str = "default") -> str:
"""
Get detailed info about a specific pod including events.
Use this when a pod is failing to understand why.
Examples:
- k8s_describe_pod("api-worker-6d4f9b", "production")
"""
return kubectl(f"describe pod {pod_name} -n {namespace}")
@tool
def k8s_get_logs(pod_name: str, namespace: str = "default", lines: int = 50) -> str:
"""
Get logs from a Kubernetes pod.
Use this to see application errors inside a pod.
Examples:
- k8s_get_logs("api-worker-6d4f9b", "production", lines=100)
"""
return kubectl(f"logs {pod_name} -n {namespace} --tail={lines}")
@tool
def k8s_get_events(namespace: str = "default") -> str:
"""
Get recent Kubernetes events in a namespace.
Sorted by time - shows warnings, errors, pod restarts.
Examples:
- k8s_get_events("production")
"""
return kubectl(f"get events -n {namespace} --sort-by='.lastTimestamp'")
@tool
def k8s_top_pods(namespace: str = "default") -> str:
"""
Get CPU and memory usage for all pods in a namespace.
Use this to find resource-hungry pods.
Examples:
- k8s_top_pods("production")
"""
return kubectl(f"top pods -n {namespace} --sort-by=memory")
Étape 4 — Outils d’information système
L’auteur, lors de l’étape 4 « Informations du système », définit les entrées, le responsable de l’étape ainsi que les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les terminations partielles silencieuses. Authentifiez-vous au niveau du gateway et réautorisez-vous au niveau du plan de données. Un token porteur seul ne constitue pas une frontière entre les tenants.
import subprocess
import psutil
from langchain_core.tools import tool
from datetime import datetime
@tool
def get_system_overview() -> str:
"""
Get a full system health snapshot:
CPU, memory, disk usage, load average, uptime.
Always call this first when diagnosing a system issue.
"""
try:
cpu = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
load = subprocess.run(
"uptime", capture_output=True, text=True
).stdout.strip()
return f"""System Overview ({datetime.now().strftime('%H:%M:%S')}):
CPU: {cpu}% used
Memory: {memory.percent}% used ({memory.used // 1024**3}GB / {memory.total // 1024**3}GB)
Disk: {disk.percent}% used ({disk.used // 1024**3}GB / {disk.total // 1024**3}GB)
Load: {load}"""
except Exception as e:
return f"Error getting system overview: {e}"
@tool
def check_service_status(service_name: str) -> str:
"""
Check if a systemd service is running and get its status.
Examples:
- check_service_status("nginx")
- check_service_status("postgresql")
- check_service_status("docker")
"""
result = subprocess.run(
f"systemctl status {service_name}",
shell=True, capture_output=True, text=True
)
return result.stdout or result.stderr
@tool
def check_port(port: int) -> str:
"""
Check what process is listening on a specific port.
Use this to verify services are bound to expected ports
or to find unexpected listeners.
Examples:
- check_port(80)
- check_port(5432)
- check_port(6379)
"""
result = subprocess.run(
f"ss -tlnp sport = :{port}",
shell=True, capture_output=True, text=True
)
if not result.stdout.strip():
return f"Nothing is listening on port {port}"
return result.stdout
@tool
def get_top_processes(sort_by: str = "cpu") -> str:
"""
Get the top 10 processes by CPU or memory usage.
Use this to find runaway processes causing system load.
Args:
sort_by: "cpu" or "memory"
"""
sort_flag = "-%cpu" if sort_by == "cpu" else "-%mem"
result = subprocess.run(
f"ps aux --sort={sort_flag} | head -11",
shell=True, capture_output=True, text=True
)
return result.stdout
@tool
def check_disk_usage(path: str = "/") -> str:
"""
Check disk usage for a path and show largest directories.
Use this when disk space is running low.
Examples:
- check_disk_usage("/")
- check_disk_usage("/var/log")
- check_disk_usage("/home")
"""
df = subprocess.run(
f"df -h {path}",
shell=True, capture_output=True, text=True
).stdout
du = subprocess.run(
f"du -sh {path}/* 2>/dev/null | sort -rh | head -10",
shell=True, capture_output=True, text=True
).stdout
return f"Disk usage for {path}:\\n{df}\\nLargest directories:\\n{du}"
Étape 5 — Construire l’agent
L’auteur indique, à l’étape 5 « Construire la scène », de définir les entrées, le responsable de l’étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Authentifiez-vous au niveau du gateway et réautorisez-vous au niveau du plan de données. Un simple jeton porteur ne constitue pas une frontière entre les tenants.
from langchain_anthropic import ChatAnthropic
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import BaseTool
from typing import List
import os
from agent.tools.bash_tools import run_command, read_file, read_logs, grep_logs
from agent.tools.kubernetes_tools import (
k8s_get_pods, k8s_describe_pod, k8s_get_logs,
k8s_get_events, k8s_top_pods
)
from agent.tools.system_tools import (
get_system_overview, check_service_status,
check_port, get_top_processes, check_disk_usage
)
SYSTEM_PROMPT = """You are an expert DevOps engineer and SRE with 10 years of experience.
You have access to tools that let you inspect systems, read logs, check services,
and query Kubernetes clusters.
Your approach:
- Start with broad system checks, then narrow down to specifics
- Always check logs when a service is misbehaving
- Look for patterns - one error often points to another
- Explain what you're doing and why as you go
- Give clear, actionable recommendations at the end
- Never run destructive commands - only read and inspect
When diagnosing issues:
1. Get system overview first if it's a general performance issue
2. Check the specific service/pod status
3. Read recent logs for errors
4. Cross-reference with system metrics
5. Provide root cause + recommended fix
You have access to these tools:
{tools}
Use this format:
Thought: [what you're thinking and why]
Action: [tool name]
Action Input: [tool input]
Observation: [what the tool returned]
... (repeat as needed)
Thought: I now have enough information to answer
Final Answer: [clear explanation + recommendations]
Begin!
Question: {input}
{agent_scratchpad}"""
def build_devops_agent(
include_kubernetes: bool = True,
verbose: bool = True
) -> AgentExecutor:
"""Build and return the DevOps AI agent."""
# Initialize Claude
llm = ChatAnthropic(
model="claude-sonnet-4-20250514",
temperature=0, # deterministic for ops tasks
max_tokens=4096,
anthropic_api_key=os.getenv("ANTHROPIC_API_KEY")
)
# Collect tools
tools: List[BaseTool] = [
run_command,
read_file,
read_logs,
grep_logs,
get_system_overview,
check_service_status,
check_port,
get_top_processes,
check_disk_usage,
]
if include_kubernetes:
tools.extend([
k8s_get_pods,
k8s_describe_pod,
k8s_get_logs,
k8s_get_events,
k8s_top_pods,
])
# Build prompt
prompt = PromptTemplate(
input_variables=["input", "tools", "tool_names", "agent_scratchpad"],
template=SYSTEM_PROMPT
)
# Create ReAct agent
agent = create_react_agent(llm, tools, prompt)
# Wrap in executor
return AgentExecutor(
agent=agent,
tools=tools,
verbose=verbose, # streams thinking to terminal
max_iterations=15, # prevent infinite loops
handle_parsing_errors=True, # recover from malformed outputs
return_intermediate_steps=True
)
Étape 6 — Le point d’entrée
import os
from dotenv import load_dotenv
from agent.agent import build_devops_agent
load_dotenv()
def run_interactive():
"""Interactive mode - chat with your agent."""
print("\\n🤖 DevOps AI Agent")
print(" Powered by Claude + LangChain")
print(" Type 'exit' to quit\\n")
print(" Examples:")
print(" → Why is nginx returning 502 errors?")
print(" → Which pod is using the most memory in production?")
print(" → Check disk space and find what's eating it")
print(" → Is the database service healthy?\\n")
agent = build_devops_agent(
include_kubernetes=True,
verbose=True
)
while True:
try:
user_input = input("\\n💬 You: ").strip()
if not user_input:
continue
if user_input.lower() in ["exit", "quit", "q"]:
print("👋 Goodbye!")
break
print("\\n🤖 Agent thinking...\\n")
result = agent.invoke({"input": user_input})
print(f"\\n{'='*60}")
print("FINAL ANSWER:")
print('='*60)
print(result["output"])
except KeyboardInterrupt:
print("\\n\\n👋 Interrupted. Goodbye!")
break
except Exception as e:
print(f"\\n❌ Agent error: {e}")
def run_single_query(query: str):
"""Run a single query - useful for scripting."""
agent = build_devops_agent(verbose=False)
result = agent.invoke({"input": query})
print(result["output"])
if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
# Single query mode
run_single_query(" ".join(sys.argv[1:]))
else:
# Interactive mode
run_interactive()
Voir en action
python main.py
💬 You: The API service is responding slowly. Find out why.
🤖 Agent thinking...
Thought: I'll start with a system overview, then check the API service.
Action: get_system_overview
Action Input: {}
Observation:
CPU: 87% used
Memory: 94% used (7.5GB / 8GB)
Disk: 45% used
Load: load average: 6.2, 5.8, 4.1
Thought: Very high CPU and memory. Check what's consuming resources.
Action: get_top_processes
Action Input: {"sort_by": "memory"}
Observation:
USER PID %CPU %MEM COMMAND
api 14821 45.2 67.3 python api_worker.py
api 14822 38.1 21.4 python api_worker.py
...
Thought: Two API worker processes are using 88% of memory combined.
Let me check the service logs for clues.
Action: read_logs
Action Input: {"service_name": "api-worker", "lines": 100}
Observation:
2026-05-22 02:31:14 WARNING Slow query: SELECT * FROM events took 8432ms
2026-05-22 02:31:19 WARNING Slow query: SELECT * FROM events took 9103ms
2026-05-22 02:31:22 ERROR Connection pool exhausted (pool_size=10)
2026-05-22 02:31:28 WARNING Slow query: SELECT * FROM events took 11201ms
Thought: DB connection pool is exhausted and queries are very slow.
Check if PostgreSQL is healthy.
Action: check_service_status
Action Input: {"service_name": "postgresql"}
Observation:
● postgresql.service - PostgreSQL RDBMS
Loaded: loaded (/lib/systemd/system/postgresql.service)
Active: active (running) since 2026-05-21 18:00:01 UTC
May 22 02:28:11 postgres[1842]: LOG: checkpoint taking longer than 30s
Thought: Checkpoints are taking too long - DB is under heavy write load.
I have enough information now.
Final Answer:
Root Cause: The API is slow because of a database bottleneck, not the app itself.
What's happening:
- PostgreSQL is performing slow checkpoints (>30s), causing all queries to queue
- The `events` table queries are taking 8-11 seconds each (should be <100ms)
- This has exhausted the connection pool (size: 10), causing all new requests to wait
Why it cascaded:
- Slow DB queries → workers hold connections longer
- Connection pool exhausts → new requests queue
- Queue builds up → high CPU from context switching → memory pressure
Immediate actions (do these now):
1. Check if the events table has a missing index:
→ psql -c "EXPLAIN ANALYZE SELECT * FROM events WHERE ..."
2. Temporarily increase connection pool:
→ Set DB_POOL_SIZE=25 and restart api-worker
3. Check for a long-running transaction blocking checkpoints:
→ psql -c "SELECT pid, age(clock_timestamp(), xact_start) FROM pg_stat_activity WHERE xact_start IS NOT NULL ORDER BY xact_start"
Root fix:
- Add an index on the events table (likely missing on timestamp or foreign key column)
- Consider query result caching for high-frequency event reads
Ajouter de la mémoire — Conversations à plusieurs tours
from langchain.memory import ConversationBufferWindowMemory
from langchain.agents import AgentExecutor, create_react_agent
# Add memory - keeps last 10 exchanges
memory = ConversationBufferWindowMemory(
k=10,
memory_key="chat_history",
return_messages=True
)
# Updated system prompt with memory
SYSTEM_PROMPT_WITH_MEMORY = """...(same as before but add)...
Previous conversation:
{chat_history}
Question: {input}
{agent_scratchpad}"""
# Pass memory to executor
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=15,
)
💬 You: What's eating disk space on the server?
🤖 Agent: [checks disk] /var/log is using 47GB.
/var/log/nginx/access.log is 38GB alone.
💬 You: How long has it been growing?
🤖 Agent: [remembers context - checks log rotation config]
Log rotation isn't configured for nginx.
The file has been growing unrotated since March 14th.
💬 You: What's the fix?
🤖 Agent: [still has full context]
Configure logrotate for nginx. Here's the exact config...
Envelopper dans une API REST
# api.py
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import uuid
from agent.agent import build_devops_agent
app = FastAPI(title="DevOps AI Agent API")
agent = build_devops_agent(verbose=False)
results_store = {} # In production: use Redis
class QueryRequest(BaseModel):
query: str
environment: str = "production"
class QueryResponse(BaseModel):
job_id: str
status: str
result: str = None
@app.post("/query", response_model=QueryResponse)
async def run_query(request: QueryRequest, background_tasks: BackgroundTasks):
"""Submit a query to the DevOps agent."""
job_id = str(uuid.uuid4())
results_store[job_id] = {"status": "running", "result": None}
def run_agent_task():
result = agent.invoke({"input": request.query})
results_store[job_id] = {
"status": "complete",
"result": result["output"]
}
background_tasks.add_task(run_agent_task)
return QueryResponse(job_id=job_id, status="running")
@app.get("/query/{job_id}", response_model=QueryResponse)
async def get_result(job_id: str):
"""Poll for query result."""
job = results_store.get(job_id)
if not job:
return QueryResponse(job_id=job_id, status="not_found")
return QueryResponse(
job_id=job_id,
status=job["status"],
result=job["result"]
)
# Run: uvicorn api:app --host 0.0.0.0 --port 8000
# From a CI/CD pipeline
curl -X POST <http://agent:8000/query> \\
-H "Content-Type: application/json" \\
-d '{"query": "Is the staging deployment healthy after this release?"}'
# From a Makefile
make check-deploy:
curl -s <http://agent:8000/query> \\
-d '{"query": "Run post-deploy checks on staging"}' | jq .result
Cas d’usage réels à essayer dès maintenant
"Nginx is returning 504 errors. Find the root cause."
"The payment service pod keeps crashing. What's happening?"
"Memory usage spiked 30 minutes ago. What caused it?"
"Run a health check on all services and give me a summary."
"Which pods in production are close to their memory limits?"
"Check if any SSL certificates are expiring in the next 30 days."
"At current growth rate, when will we run out of disk space in /var/log?"
"Which services are consistently above 80% CPU?"
"Find any processes listening on unexpected ports."
"Check auth logs for failed login attempts in the last hour."
"Are there any world-writable files in /etc?"
Erreurs courantes et solutions
# Set a hard iteration limit
AgentExecutor(max_iterations=15, ...) # never go above 20
# Always truncate in your tools
if len(output) > 3000:
output = output[:3000] + "\\n... [truncated]"
# Don't rely on the LLM to self-police
# Enforce safety in the tool code itself (like we did in safety.py)
# The LLM cannot bypass code-level blocks
# Use streaming to show progress
async for chunk in agent.astream({"input": query}):
if "actions" in chunk:
for action in chunk["actions"]:
print(f" → Running: {action.tool}({action.tool_input})")