AgentOps Production Implementation Guide 2026

from pydantic import BaseModel, Field
from typing import List, Literal

class ToolConfig(BaseModel):
    name: str
    api_version: str
    endpoint: str
    timeout_ms: int = 5000

class AgentConfig(BaseModel):
    agent_version: str = Field(..., pattern=r"^\d+\.\d+\.\d+$")
    system_prompt: str
    model: Literal["gpt-5.2-turbo", "claude-opus-4.5", "gemini-3-pro"]
    tools: List[ToolConfig]
    max_iterations: int = 10
    temperature: float = 0.7

# Version control this file in Git
supervisor_v2_1_0 = AgentConfig(
    agent_version="2.1.0",
    system_prompt="You are a supervisor agent coordinating research and synthesis...",
    model="claude-opus-4.5",
    tools=[
        ToolConfig(name="search", api_version="v2", endpoint="https://api.search.internal"),
        ToolConfig(name="sql_query", api_version="v1", endpoint="https://db.internal"),
    ],
    max_iterations=15
)

async def handle_request(user_input: str):
    # Production agent (blocking)
    result_prod = await agent_v1_5.run(user_input)

    # Shadow agent (non-blocking, fire-and-forget)
    asyncio.create_task(agent_v2_0.run(user_input, shadow_mode=True))

    return result_prod  # User sees only v1.5 output

async def execute_tool_with_approval(tool_name: str, params: dict, risk_level: str):
    if risk_level == "high":
        # Pause execution, request human approval
        approval = await request_human_approval(
            message=f"Agent wants to call {tool_name} with {params}. Approve?",
            timeout_seconds=300
        )
        if not approval:
            raise ToolExecutionDenied(f"Human denied {tool_name} call")

    return await execute_tool(tool_name, params)

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Any
import asyncio
import uuid
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import logging

# Initialize OpenTelemetry
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
span_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(span_exporter))

# Prometheus metrics
agent_tasks_total = Counter('agent_tasks_total', 'Total agent tasks', ['agent_id', 'status'])
agent_task_duration = Histogram('agent_task_duration_seconds', 'Agent task duration', ['agent_id'])
agent_cost_total = Counter('agent_cost_total', 'Total agent cost USD', ['agent_id', 'user_id'])
agent_steps_total = Counter('agent_steps_total', 'Total agent steps', ['agent_id'])
agent_active_tasks = Gauge('agent_active_tasks', 'Currently executing tasks', ['agent_id'])

logger = logging.getLogger(__name__)

@dataclass
class AgentVersion:
    """Semantic versioning for agents"""
    major: int
    minor: int
    patch: int

    def __str__(self) -> str:
        return f"{self.major}.{self.minor}.{self.patch}"

    @classmethod
    def from_string(cls, version: str) -> 'AgentVersion':
        parts = version.split('.')
        return cls(int(parts[0]), int(parts[1]), int(parts[2]))

@dataclass
class CostBreakdown:
    """Cost attribution per agent execution"""
    input_tokens: int = 0
    output_tokens: int = 0
    cached_tokens: int = 0
    tool_calls: int = 0

    def calculate_cost(self, model: str) -> float:
        """Calculate cost based on model pricing (GPT-5.2 example)"""
        costs = {
            "gpt-5.2": {"input": 3.0, "output": 15.0, "cached": 1.5},  # per 1M tokens
            "claude-opus-4.5": {"input": 15.0, "output": 75.0, "cached": 7.5},
            "gemini-3-pro": {"input": 1.25, "output": 5.0, "cached": 0.625}
        }
        pricing = costs.get(model, costs["gpt-5.2"])

        input_cost = (self.input_tokens - self.cached_tokens) * pricing["input"] / 1_000_000
        cached_cost = self.cached_tokens * pricing["cached"] / 1_000_000
        output_cost = self.output_tokens * pricing["output"] / 1_000_000
        tool_cost = self.tool_calls * 0.001  # $0.001 per tool call

        return input_cost + cached_cost + output_cost + tool_cost

@dataclass
class AuditLogEntry:
    """Compliance audit logging"""
    timestamp: datetime
    task_id: str
    agent_id: str
    agent_version: str
    user_id: str
    action_type: str  # "tool_call", "llm_decision", "human_approval"
    action_details: Dict[str, Any]
    metadata: Dict[str, Any] = field(default_factory=dict)

class AgentOpsMonitor:
    """
    Production AgentOps monitoring wrapper for LangGraph agents.

    Implements:
    - Semantic versioning
    - Distributed tracing with OpenTelemetry
    - Cost tracking and anomaly detection
    - Audit logging for compliance
    - Shadow deployment for A/B testing
    - Prometheus metrics export
    """

    def __init__(
        self,
        agent_id: str,
        agent_version: AgentVersion,
        model: str,
        cost_threshold_usd: float = 1.0
    ):
        self.agent_id = agent_id
        self.version = agent_version
        self.model = model
        self.cost_threshold = cost_threshold_usd
        self.audit_log: List[AuditLogEntry] = []

    async def execute_task(
        self,
        task_input: str,
        user_id: str,
        metadata: Optional[Dict] = None
    ) -> Dict[str, Any]:
        """
        Execute agent task with full monitoring.

        Returns:
            Dict with keys: result, cost, trace_id, completion_status
        """
        task_id = str(uuid.uuid4())
        metadata = metadata or {}

        # Start distributed trace
        with tracer.start_as_current_span(
            f"agent_task_{self.agent_id}",
            attributes={
                "agent.id": self.agent_id,
                "agent.version": str(self.version),
                "task.id": task_id,
                "user.id": user_id,
            }
        ) as span:
            start_time = datetime.utcnow()
            agent_active_tasks.labels(agent_id=self.agent_id).inc()

            try:
                # Execute agent with cost tracking
                result, cost = await self._execute_with_cost_tracking(
                    task_id, task_input, user_id, metadata
                )

                # Check for cost anomaly
                if cost.calculate_cost(self.model) > self.cost_threshold:
                    logger.warning(
                        f"Cost anomaly: Task {task_id} cost ${cost.calculate_cost(self.model):.3f}, "
                        f"threshold ${self.cost_threshold}"
                    )
                    await self._trigger_cost_alert(task_id, cost, user_id)

                # Record success
                duration = (datetime.utcnow() - start_time).total_seconds()
                agent_tasks_total.labels(agent_id=self.agent_id, status="success").inc()
                agent_task_duration.labels(agent_id=self.agent_id).observe(duration)
                agent_cost_total.labels(agent_id=self.agent_id, user_id=user_id).inc(
                    cost.calculate_cost(self.model)
                )

                span.set_attribute("task.status", "success")
                span.set_attribute("task.cost_usd", cost.calculate_cost(self.model))

                return {
                    "result": result,
                    "cost": cost.calculate_cost(self.model),
                    "trace_id": span.get_span_context().trace_id,
                    "completion_status": "success"
                }

            except Exception as e:
                # Record failure
                agent_tasks_total.labels(agent_id=self.agent_id, status="failure").inc()
                span.set_attribute("task.status", "failure")
                span.set_attribute("error.message", str(e))

                logger.error(f"Agent task {task_id} failed: {e}")

                # Audit log failure
                self._log_audit(
                    task_id, user_id, "task_failure",
                    {"error": str(e), "input": task_input}
                )

                raise

            finally:
                agent_active_tasks.labels(agent_id=self.agent_id).dec()

    async def _execute_with_cost_tracking(
        self,
        task_id: str,
        task_input: str,
        user_id: str,
        metadata: Dict
    ) -> tuple[str, CostBreakdown]:
        """Execute agent and track token/tool costs"""
        cost = CostBreakdown()

        # Simulate agent execution (replace with actual LangGraph agent call)
        with tracer.start_as_current_span("agent_reasoning"):
            # Step 1: Initial LLM call
            result_step1 = await self._llm_call(
                prompt=f"Analyze task: {task_input}",
                user_id=user_id,
                task_id=task_id
            )
            cost.input_tokens += 120
            cost.output_tokens += 45
            agent_steps_total.labels(agent_id=self.agent_id).inc()

            # Step 2: Tool call (if needed)
            if "search" in result_step1.lower():
                with tracer.start_as_current_span("tool_call_search"):
                    await self._tool_call("web_search", {"query": task_input}, task_id, user_id)
                    cost.tool_calls += 1
                    agent_steps_total.labels(agent_id=self.agent_id).inc()

            # Step 3: Final synthesis
            result_final = await self._llm_call(
                prompt=f"Synthesize: {result_step1}",
                user_id=user_id,
                task_id=task_id,
                use_cache=True
            )
            cost.input_tokens += 200
            cost.cached_tokens += 120  # System prompt cached
            cost.output_tokens += 180
            agent_steps_total.labels(agent_id=self.agent_id).inc()

        return result_final, cost

    async def _llm_call(
        self,
        prompt: str,
        user_id: str,
        task_id: str,
        use_cache: bool = False
    ) -> str:
        """Make LLM call with audit logging"""
        with tracer.start_as_current_span(
            "llm_call",
            attributes={"model": self.model, "cache_enabled": use_cache}
        ):
            # Audit log the decision
            self._log_audit(
                task_id, user_id, "llm_decision",
                {"model": self.model, "prompt_preview": prompt[:100]}
            )

            # Simulate LLM call
            await asyncio.sleep(0.1)  # Replace with actual LLM API call
            return f"LLM response to: {prompt[:50]}..."

    async def _tool_call(
        self,
        tool_name: str,
        params: Dict,
        task_id: str,
        user_id: str
    ) -> Any:
        """Execute tool with audit logging"""
        # Audit log tool call
        self._log_audit(
            task_id, user_id, "tool_call",
            {"tool": tool_name, "params": params}
        )

        # Simulate tool execution
        await asyncio.sleep(0.05)
        return {"status": "success", "data": "tool result"}

    def _log_audit(
        self,
        task_id: str,
        user_id: str,
        action_type: str,
        action_details: Dict[str, Any]
    ):
        """Write compliance audit log entry"""
        entry = AuditLogEntry(
            timestamp=datetime.utcnow(),
            task_id=task_id,
            agent_id=self.agent_id,
            agent_version=str(self.version),
            user_id=user_id,
            action_type=action_type,
            action_details=action_details
        )
        self.audit_log.append(entry)

        # In production: Write to tamper-proof audit database
        # await audit_db.write(entry)

    async def _trigger_cost_alert(
        self,
        task_id: str,
        cost: CostBreakdown,
        user_id: str
    ):
        """Alert on cost anomaly"""
        logger.warning(
            f"COST ALERT: Task {task_id} for user {user_id} cost "
            f"${cost.calculate_cost(self.model):.3f} (threshold: ${self.cost_threshold})"
        )
        # In production: Send to alerting system (PagerDuty, Slack, etc.)

    async def shadow_deploy(
        self,
        production_agent: 'AgentOpsMonitor',
        task_input: str,
        user_id: str
    ) -> Dict[str, Any]:
        """
        Run this agent in shadow mode alongside production agent.
        Production result returned to user, shadow result logged for comparison.
        """
        # Execute both in parallel
        prod_task = production_agent.execute_task(task_input, user_id)
        shadow_task = self.execute_task(task_input, user_id, metadata={"shadow_mode": True})

        prod_result, shadow_result = await asyncio.gather(
            prod_task, shadow_task, return_exceptions=True
        )

        # Compare results (for offline evaluation)
        if not isinstance(shadow_result, Exception):
            logger.info(
                f"Shadow comparison - Prod cost: ${prod_result['cost']:.3f}, "
                f"Shadow cost: ${shadow_result['cost']:.3f}"
            )
            # In production: Log to evaluation database for analysis

        return prod_result  # Return only production result to user

    def export_audit_log(self, start_time: datetime, end_time: datetime) -> List[AuditLogEntry]:
        """Export audit logs for compliance review"""
        return [
            entry for entry in self.audit_log
            if start_time <= entry.timestamp <= end_time
        ]

# Example usage
async def main():
    # Start Prometheus metrics server
    start_http_server(8000)

    # Initialize production and shadow agents
    agent_prod = AgentOpsMonitor(
        agent_id="supervisor",
        agent_version=AgentVersion(2, 1, 0),
        model="claude-opus-4.5",
        cost_threshold_usd=0.50
    )

    agent_shadow = AgentOpsMonitor(
        agent_id="supervisor",
        agent_version=AgentVersion(2, 2, 0),  # Testing new version
        model="claude-opus-4.5",
        cost_threshold_usd=0.50
    )

    # Execute task with shadow deployment
    result = await agent_shadow.shadow_deploy(
        production_agent=agent_prod,
        task_input="Analyze Q4 2025 sales data and identify growth opportunities",
        user_id="user_12345"
    )

    print(f"Task result: {result['result']}")
    print(f"Cost: ${result['cost']:.3f}")
    print(f"Trace ID: {result['trace_id']}")

    # Export audit logs
    audit_entries = agent_prod.export_audit_log(
        start_time=datetime(2026, 2, 1),
        end_time=datetime(2026, 2, 28)
    )
    print(f"Exported {len(audit_entries)} audit log entries for compliance")

if __name__ == "__main__":
    asyncio.run(main())