Notas prácticas: Crea tu primer agente de IA para DevOps utilizando LangChain y herramientas Bash
Guía paso a paso práctica: Crea tu primer agente de IA DevOps utilizando LangChain y herramientas Bash: contratos, verificaciones y espacios para código listos para usar para los equipos que implementan este patrón.
Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: Construir tu primer agente de IA DevOps utilizando LangChain + herramientas Bash. El enfoque está en pasos operativos claros, verificaciones explícitas y código que se puede incorporar directamente a un repositorio sin necesidad de adivinar su propósito. El autor debe definir la etapa general, los datos de entrada, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Se debe documentar tanto el camino óptimo como el de recuperación. Las intentonas repetidas, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.
¿Qué estamos construyendo realmente?
Al trabajar en la fase de “¿Qué estamos construyendo?”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Registre el nombre de la herramienta, el hash de los argumentos, la latencia y el resultado de cada llamada. Depurar bucles sin esa huella desperdicia horas.
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."
¿Por qué LangChain para esto?
Al trabajar en la etapa “¿Por qué LangChain?”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Registre el nombre de la herramienta, el hash de los argumentos, la latencia y el resultado de cada llamada. Depurar bucles de agentes sin ese registro desperdicia horas.
Requisitos previos
Al trabajar en la etapa de Requisitos previos, anote primero el contrato: los datos de entrada necesarios, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la fase de demostración a entornos compartidos. Registre el nombre de la herramienta, el hash de los argumentos, la latencia y el resultado de cada llamada. Sin ese registro, depurar bucles del agente consume horas innecesarias.
# 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
Al trabajar en la etapa de Requisitos previos, anote primero el contrato: los datos de entrada necesarios, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación garantiza que los cambios posteriores en el código sean transparentes. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras adicionales realizadas posteriormente.
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
Bucle del Agente ReAct
La etapa del bucle de agente ReAct funciona mejor cuando se trata como una superficie medible. Capture un transcripte exitoso, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Exponga herramientas con esquemas limitados y etiquetas explícitas de efectos secundarios. Los hosts necesitan saber qué llamadas modifican el estado antes de aprobarlas automáticamente.
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..." │
└─────────────────────────────────────────────┘
Paso 1: La seguridad primero
La etapa 1, “Seguridad primero”, funciona mejor cuando se trata como una superficie medible. Capture un registro de éxito ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Exponga herramientas con esquemas limitados y etiquetas explícitas de efectos secundarios. Los hosts necesitan saber qué llamadas modifican el estado antes de aprobarlas automáticamente.
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"
Etapa 2: Herramientas básicas de Bash
La etapa Step 2 Core Bash funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la demostración a entornos compartidos. Exponga herramientas con esquemas limitados y etiquetas explícitas de efectos secundarios. Los administradores necesitan saber qué llamadas modifican el estado antes de aprobarlas automáticamente. La etapa Step 2 Core Bash funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente tanto el camino óptimo como el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.
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)}"
Paso 3: Herramientas de Kubernetes
En la etapa de herramientas de Kubernetes del Paso 3, el autor debe definir las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Es preferible utilizar unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complejo e entrelazado. Se debe autenticar en la pasarela y volver a autorizarlo en el plano de datos; un token portador por sí solo no constituye un límite entre tenencias.
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")
Paso 4 — Herramientas de información del sistema
En la etapa 4 del sistema, el autor debe definir las entradas, el responsable de dicha etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace cualquier completación parcial silenciosa. Autentíquese en la pasarela y vuelva a autorizarse en el plano de datos; un token portador por sí solo no constituye un límite entre tenencias.
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}"
Paso 5: Crear el agente
El autor indica que en el Paso 5, antes de modificar el código, se debe construir la etapa, definir las entradas, identificar al responsable de dicha etapa y establecer los criterios de finalización. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido, sin tener que adivinar el estado oculto. Se deben registrar los tiempos de ejecución y el costo de los tokens o consultas junto con los resultados funcionales. Tener visibilidad sobre los costos desde el principio evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Es necesario autenticarse en la pasarela y volver a autorizarlo en el plano de datos; un token portador por sí solo no constituye un límite entre tenencias.
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
)
Paso 6: El punto de entrada
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()
Véalo en acción
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
Añadir memoria: conversaciones de múltiples turnos
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...
Envolverlo como una 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
Casos de uso reales para probar ahora mismo
"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?"
Errores comunes y soluciones
# 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})")