Building 24/7 Autonomous Agents with LangGraph and NIM

[ Traditional Linear Chain (Fragile) ]
Input ──▶ [ LLM Call ] ──▶ [ Tool Execution ] ──▶ [ Unhandled Error / Hallucination ] ──▶ CRASH

[ Cyclic StateGraph Daemon (Resilient & Self-Healing) ]
Input ──▶ [ DeepSeek-R1 Planner ] ◄───────────────────────────┐
                 │                                            │
                 ▼                                            │ (Reflection & Correction Loop)
          [ Hermes-3 Executor ] ──▶ [ Critic / Evaluator ] ───┘
                 │                         ▲
                 ▼                         │
          [ MCP Gateway Server ] ──────────┘
                 │
                 ▼ (Verified Trajectory)
          [ Atomic State Checkpoint (PostgreSQL / SQLite) ]

import os
import json
import logging
from typing import Dict, Any, List, Optional
from openai import OpenAI

logger = logging.getLogger("NIMHybridClient")

class NIMHybridClient:
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("NVIDIA_API_KEY")
        if not self.api_key:
            raise ValueError("NVIDIA_API_KEY must be provided or set in environment.")

        self.client = OpenAI(
            base_url="https://integrate.api.nvidia.com/v1",
            api_key=self.api_key,
            timeout=45.0
        )
        self.planner_model = "deepseek-ai/deepseek-r1"
        self.executor_model = "nousresearch/hermes-3-llama-3.1-70b"

    def plan_with_r1(self, goal: str, context: str) -> str:
        """Invokes DeepSeek-R1 for deep reasoning and task decomposition."""
        messages = [
            {"role": "system", "content": "You are a Chief Systems Architect. Formulate an optimal, verifiable DAG plan for the objective."},
            {"role": "user", "content": f"Objective: {goal}\nContext:\n{context}"}
        ]
        response = self.client.chat.completions.create(
            model=self.planner_model,
            messages=messages,
            temperature=0.6,
            max_tokens=2048
        )
        return response.choices[0].message.content

    def execute_with_hermes(self, messages: List[Dict[str, str]], tools: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
        """Invokes Hermes-3 for deterministic tool calling and schema compliance."""
        kwargs = {
            "model": self.executor_model,
            "messages": messages,
            "temperature": 0.1,
            "max_tokens": 1024
        }
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = "auto"

        response = self.client.chat.completions.create(**kwargs)
        choice = response.choices[0]
        message = choice.message

        return {
            "content": message.content or "",
            "tool_calls": [
                {
                    "id": tc.id,
                    "function": {
                        "name": tc.function.name,
                        "arguments": json.loads(tc.function.arguments)
                    }
                }
                for tc in (message.tool_calls or [])
            ]
        }

import operator
from typing import Annotated, List, Dict, Any, TypedDict, Literal
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver

# 1. State Definition
class AgentDaemonState(TypedDict):
    task_id: str
    goal: str
    plan: List[str]
    current_step_index: int
    execution_history: Annotated[List[Dict[str, Any]], operator.add]
    pending_tool_calls: List[Dict[str, Any]]
    latest_output: str
    evaluation_score: float
    critic_feedback: str
    retry_count: int
    is_complete: bool

nim_client = NIMHybridClient()

def planner_node(state: AgentDaemonState) -> Dict[str, Any]:
    context = f"Critic Feedback from Previous Attempt: {state.get('critic_feedback', 'None')}"
    plan_raw = nim_client.plan_with_r1(state["goal"], context)
    steps = [line.strip("- *0123456789. ") for line in plan_raw.split("\n") if len(line.strip()) > 5][:6]
    return {"plan": steps, "current_step_index": 0, "retry_count": state.get("retry_count", 0) + 1, "critic_feedback": ""}

def executor_node(state: AgentDaemonState) -> Dict[str, Any]:
    step = state["plan"][state["current_step_index"]]
    messages = [
        {"role": "system", "content": "You are an Autonomous Systems Executor. Execute the task step precisely using tools."},
        {"role": "user", "content": f"Target Step: {step}\nRecent History:\n{json.dumps(state['execution_history'][-2:])}"}
    ]
    tools = [
        {
            "type": "function",
            "function": {
                "name": "system_telemetry",
                "description": "Retrieves real-time CPU, RAM, and GPU cluster metrics.",
                "parameters": {"type": "object", "properties": {}, "required": []}
            }
        }
    ]
    result = nim_client.execute_with_hermes(messages, tools=tools)
    return {
        "latest_output": result["content"],
        "pending_tool_calls": result["tool_calls"],
        "execution_history": [{"step": step, "output": result["content"], "tool_calls": result["tool_calls"]}]
    }

def tool_node(state: AgentDaemonState) -> Dict[str, Any]:
    results = []
    for tc in state["pending_tool_calls"]:
        func = tc["function"]["name"]
        if func == "system_telemetry":
            res = {"cpu_usage_pct": 18.5, "vram_free_gb": 19.2, "gpu_temp_c": 52}
        else: 
            res = {"status": "ok", "message": f"Executed {func}"}
        results.append({"id": tc["id"], "name": func, "response": res})
    return {"pending_tool_calls": [], "execution_history": [{"tool_responses": results}]}

def evaluator_node(state: AgentDaemonState) -> Dict[str, Any]:
    step = state["plan"][state["current_step_index"]]
    judge_prompt = [
        {"role": "system", "content": "You are a Quality Arbiter. Score the execution (0.0 to 1.0) and return JSON: {\"score\": float, \"feedback\": str}"},
        {"role": "user", "content": f"Goal: {state['goal']}\nStep: {step}\nOutput: {state['latest_output']}"}
    ]
    res = nim_client.execute_with_hermes(judge_prompt)
    try:
        data = json.loads(res["content"])
        score = float(data.get("score", 0.0))
        feedback = data.get("feedback", "")
    except Exception:
        score = 0.5
        feedback = "Evaluator failed JSON parse."
    return {"evaluation_score": score, "critic_feedback": feedback}

def route_after_executor(state: AgentDaemonState) -> Literal["tools", "evaluator"]:
    return "tools" if state["pending_tool_calls"] else "evaluator"

def route_after_evaluator(state: AgentDaemonState) -> Literal["advance", "retry", "done", "failed"]:
    if state["evaluation_score"] >= 0.85:
        if state["current_step_index"] + 1 < len(state["plan"]):
            return "advance"
        return "done"
    if state["retry_count"] >= 4:
        return "failed"
    return "retry"

def advance_step(state: AgentDaemonState) -> Dict[str, Any]:
    return {"current_step_index": state["current_step_index"] + 1}

# Constructing the Graph
builder = StateGraph(AgentDaemonState)
builder.add_node("planner", planner_node)
builder.add_node("executor", executor_node)
builder.add_node("tools", tool_node)
builder.add_node("evaluator", evaluator_node)
builder.add_node("advance", advance_step)

builder.set_entry_point("planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", route_after_executor, {"tools": "tools", "evaluator": "evaluator"})
builder.add_edge("tools", "executor")
builder.add_conditional_edges("evaluator", route_after_evaluator, {
    "advance": "advance",
    "retry": "planner",
    "done": END,
    "failed": END
})
builder.add_edge("advance", "executor")

checkpointer = SqliteSaver.from_conn_string("agent_daemon_state.db")
agent_app = builder.compile(checkpointer=checkpointer)

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("EisenEngineClusterGateway")

@mcp.tool()
def get_vulkan_pipeline_metrics() -> dict:
    """Returns draw calls, frame times, and GPU memory allocations from active game engine nodes."""
    return {
        "draw_calls_per_frame": 62,
        "avg_frame_time_ms": 16.2,
        "p99_frame_time_ms": 17.8,
        "vram_in_use_mb": 512.4,
        "active_entities": 10240
    }

if __name__ == "__main__":
    mcp.run()

import dataclasses

@dataclasses.dataclass
class TrajectoryEvaluationResult:
    pci: float          # Plan Coherence Index
    tsep: float         # Tool Selection & Execution Precision
    stiv: float         # State Transition Invariant Validity
    fghs: float         # Factual Grounding & Hallucination Suppression
    composite: float    # Weighted Overall Score
    passed_gate: bool

class TrajectoryEvaluator:
    def __init__(self, judge_client: NIMHybridClient):
        self.judge = judge_client

    def score_trajectory(self, trajectory: List[Dict[str, Any]], goal: str) -> TrajectoryEvaluationResult:
        # 1. State Transition Invariant Validity (STIV)
        illegal_transitions = sum(1 for i in range(len(trajectory)-1) 
                                  if trajectory[i].get("node") == "executor" 
                                  and trajectory[i+1].get("node") not in ["tools", "evaluator"])
        stiv = max(0.0, 1.0 - (illegal_transitions / max(1, len(trajectory)-1)))

        # 2. Tool Selection & Execution Precision (TSEP)
        tool_steps = [s for s in trajectory if "tool_responses" in s]
        tsep = 1.0 if not tool_steps else max(0.0, 1.0 - (sum(1 for t in tool_steps if "error" in str(t).lower()) / len(tool_steps)))

        # 3. LLM-Judged Metrics (PCI & FGHS)
        prompt = [
            {"role": "system", "content": "Score Plan Coherence (PCI) and Factual Grounding (FGHS) from 0.0 to 1.0. Return JSON: {\"pci\": float, \"fghs\": float}"},
            {"role": "user", "content": f"Goal: {goal}\nTrajectory:\n{json.dumps(trajectory)}"}
        ]
        res = self.judge.execute_with_hermes(prompt)
        try:
            d = json.loads(res["content"])
            pci, fghs = float(d.get("pci", 0.85)), float(d.get("fghs", 0.90))
        except Exception:
            pci, fghs = 0.80, 0.80

        composite = (0.25 * pci) + (0.30 * tsep) + (0.25 * stiv) + (0.20 * fghs)

        return TrajectoryEvaluationResult(
            pci=pci, tsep=tsep, stiv=stiv, fghs=fghs,
            composite=composite, passed_gate=(composite >= 0.88)
        )

[ Agent Enters Critical Node (e.g. DB Migration / Deploy) ]
                          │
                          ▼
            [ LangGraph interrupt() Triggered ]
                          │
         (State persisted to Postgres / SQLite)
                          │
                          ▼
        [ Outbound Webhook to Slack / Discord Bot ]
        "Approval Required: Task #812 requires DB write.
         [ ✅ APPROVE ]    [ ❌ REJECT ]    [ 💬 GUIDANCE ]"
                          │
               (Human clicks button in Slack)
                          │
                          ▼
        [ Webhook Receiver -> agent_app.invoke(Command(resume=...)) ]
                          │
                          ▼
        [ Agent Awakens and Continues Cleanly ]

import time
import logging

logger = logging.getLogger("CircuitBreaker")

class BudgetExhaustedException(Exception):
    pass

class CostCircuitBreaker:
    def __init__(self, max_cost_per_session_usd: float = 1.50, max_token_velocity_per_min: int = 50000):
        self.max_cost_usd = max_cost_per_session_usd
        self.max_token_velocity = max_token_velocity_per_min
        self.session_cost_usd = 0.0
        self.token_history = []  # List of (timestamp, token_count)

    def record_usage(self, prompt_tokens: int, completion_tokens: int, model: str):
        # Pricing approximations per 1M tokens (e.g. 70B model)
        cost = (prompt_tokens * 0.0000008) + (completion_tokens * 0.0000025)
        self.session_cost_usd += cost
        now = time.time()
        self.token_history.append((now, prompt_tokens + completion_tokens))

        # Check budget limit
        if self.session_cost_usd >= self.max_cost_usd:
            logger.critical(f"🚨 CIRCUIT BREAKER TRIPPED: Session cost ${self.session_cost_usd:.4f} exceeded limit ${self.max_cost_usd:.2f}!")
            raise BudgetExhaustedException(f"Spend limit exceeded: ${self.session_cost_usd:.4f}")

        # Check token velocity (sliding 60-second window)
        cutoff = now - 60.0
        self.token_history = [(ts, cnt) for ts, cnt in self.token_history if ts >= cutoff]
        rolling_tokens = sum(cnt for _, cnt in self.token_history)

        if rolling_tokens > self.max_token_velocity:
            logger.warning(f"⚠️ Token velocity spike: {rolling_tokens} TPM. Throttling worker...")
            time.sleep(2.0)

from langgraph.types import interrupt, Command
import requests

def high_risk_approval_node(state: AgentDaemonState) -> Dict[str, Any]:
    """Suspends graph execution and triggers a Slack webhook for manual approval."""
    payload = {
        "text": f"🚨 *Agent Approval Required* (Task `{state['task_id']}`)\n*Action:* {state['latest_output']}\n*Score:* {state['evaluation_score']}",
        "attachments": [{
            "text": "Approve action to proceed?",
            "fallback": "Cannot approve on this client",
            "callback_id": state["task_id"],
            "actions": [
                {"name": "decision", "text": "Approve", "type": "button", "value": "approve"},
                {"name": "decision", "text": "Reject", "type": "button", "value": "reject", "style": "danger"}
            ]
        }]
    }

    # Send non-blocking webhook to Slack / Discord
    slack_webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if slack_webhook_url:
        try:
            requests.post(slack_webhook_url, json=payload, timeout=5.0)
        except Exception as e:
            logger.error(f"Failed to send Slack webhook: {e}")

    # NATIVE INTERRUPT: Saves state to checkpointer and halts execution
    human_decision = interrupt({
        "question": "Do you approve this deployment action?",
        "task_id": state["task_id"]
    })

    if human_decision.get("decision") == "approve":
        return {"critic_feedback": "Approved by human supervisor."}
    else:
        return {"critic_feedback": f"Rejected by human: {human_decision.get('reason', 'Denied')}"}

import redis
import json

class AgentFleetQueue:
    def __init__(self, stream_key: str = "agent_tasks_stream", group_name: str = "agent_fleet_workers"):
        self.r = redis.Redis(host="localhost", port=6379, db=0)
        self.stream = stream_key
        self.group = group_name
        try:
            self.r.xgroup_create(self.stream, self.group, id="0", mkstream=True)
        except redis.exceptions.ResponseError:
            pass  # Group already exists

    def push_task(self, task_id: str, goal: str):
        self.r.xadd(self.stream, {"task_id": task_id, "goal": goal})

    def consume_task(self, worker_id: str, block_ms: int = 5000):
        # Read unique task assigned to this worker in the consumer group
        messages = self.r.xreadgroup(self.group, worker_id, {self.stream: ">"}, count=1, block=block_ms)
        if not messages:
            return None
        msg_id, data = messages[0][1][0]
        return msg_id, {k.decode(): v.decode() for k, v in data.items()}

    def acknowledge_task(self, msg_id: str):
        self.r.xack(self.stream, self.group, msg_id)

[ Production Daemon Fleet (405B / 70B Teacher) ]
                        │
                        ▼
       [ 4-D Trajectory Evaluation Harness ]
                        │
             (Composite Score >= 0.92?)
             /                        \
          [YES]                      [NO]
           /                            \
          v                              v
  [ Dataset Curation ]           [ Dead-Letter Queue (DLQ) ]
  (Sanitize & Format)            (Root-Cause Failure Debug)
          │
          ▼
  [ Group Relative Policy Optimization (GRPO) ]
          │
          ▼
  [ Export 4-bit Quantized Model to Local Edge Workers ]

import asyncio
import logging
import signal
from typing import NoReturn

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("247AgentDaemon")

class ProductionAgentWorker:
    def __init__(self):
        self.is_running = True
        self.breaker = CostCircuitBreaker(max_cost_per_session_usd=2.00)
        self._setup_signals()

    def _setup_signals(self):
        signal.signal(signal.SIGINT, self._handle_shutdown)
        signal.signal(signal.SIGTERM, self._handle_shutdown)

    def _handle_shutdown(self, signum, frame):
        logger.warning(f"Termination signal received ({signum}). Draining agent worker queue...")
        self.is_running = False

    async def start(self) -> NoReturn:
        logger.info("🚀 Production 24/7 Agent Daemon Initialized.")
        backoff_delay = 2.0

        while self.is_running:
            try:
                task_id = f"task_{int(asyncio.get_event_loop().time() * 1000)}"
                logger.info(f"Processing Task ID: {task_id}")

                config = {"configurable": {"thread_id": task_id}}
                initial_state = {
                    "task_id": task_id,
                    "goal": "Audit GPU memory pressure and optimize ECS batch dispatching.",
                    "execution_history": [],
                    "retry_count": 0
                }

                # Execute the LangGraph workflow asynchronously in a thread pool
                result = await asyncio.to_thread(agent_app.invoke, initial_state, config=config)
                logger.info(f"✅ Completed Task {task_id} with Score: {result.get('evaluation_score', 1.0)}")

                backoff_delay = 2.0
                await asyncio.sleep(10.0)

            except BudgetExhaustedException:
                logger.critical("🛑 Shutting down daemon due to budget circuit breaker.")
                break
            except Exception as exc:
                logger.error(f"❌ Daemon error: {exc}", exc_info=True)
                await asyncio.sleep(backoff_delay)
                backoff_delay = min(60.0, backoff_delay * 2.0)

if __name__ == "__main__":
    worker = ProductionAgentWorker()
    asyncio.run(worker.start())