Articles/Agents & Manager

◈ Agents & Manager/2026-03-14Advanced

LangChain and Antigravity Pipelines, Designed Backwards from Permissions, Failures, and Cost

Design pipelines starting from where they resume after a crash, not from how the chains connect. With permission tiers, checkpointing, and measured before-and-after numbers from my own runs.

LangChain³ AI Pipelines Architecture⁵ Production¹⁰ Chain Composition Memory Management

✦ Premium Article

The first time I ran an asset-classification pipeline overnight for one of my wallpaper apps, I woke up to a log that simply stopped. Roughly half of a few thousand images had been processed. Rate limiting was the trigger, but that wasn't the real problem. The real problem was that nothing had recorded how far it got.

So I ran the whole thing again, and paid twice for the same images.

Since then, I decide one thing before I decide how the chains connect: where does this pipeline resume when it dies halfway through. LangChain supplies the composable components. Antigravity supplies the orchestration layer that runs them alongside a human. What sits between them are three decisions this article is really about — the permission boundary, the record of failure, and the visibility of cost.

We'll walk through composable chain design, memory management, routing, and resilience, then finish in the territory the official docs tend to skip: permission tiers and cost measured from real runs.

Understanding AI Pipeline Architecture

The Pipeline Abstraction

A pipeline isn't simply a chain of LLM calls. It's a composition of specialized components working in concert:

Input → Preprocessing → Routing → Chain Selection → Execution → Post-processing → Output
           ↓                ↓              ↓              ↓              ↓
      Data Validation   Context        Multi-branch   Logging      Response
      Transformation    Retrieval      Fallbacks      Monitoring   Formatting
                        Caching        Error Handler  Metrics

Why this matters: Generic pipeline designs break under production load. Custom architectures distribute complexity intelligently—moving expensive operations (like retrieval) outside the critical path, caching aggressively, and gracefully degrading when components fail.

Five Pipeline Patterns

LangChain enables five distinct patterns, each suited to different problems:

Sequential Pipelines — Linear chains of operations (document Q&A, summarization)
Branching Pipelines — Conditional routing based on input characteristics
Recursive Pipelines — Chains that call themselves for refinement or decomposition
Parallel Pipelines — Multiple chains executing simultaneously, results merged
Adaptive Pipelines — Runtime behavior determined by metrics, performance data, or state

Understanding when to apply each pattern dramatically improves system efficiency and reliability.

Building Sequential Pipelines with Composable Components

Core Architecture

A production sequential pipeline separates concerns into discrete, testable layers:

# config/pipeline_config.py
from dataclasses import dataclass
from typing import Optional, Dict, Any
from enum import Enum
 
class ChainStage(Enum):
    """Pipeline execution stages"""
    INPUT_VALIDATION = "input_validation"
    CONTEXT_RETRIEVAL = "context_retrieval"
    PROMPT_CONSTRUCTION = "prompt_construction"
    LLM_EXECUTION = "llm_execution"
    OUTPUT_PARSING = "output_parsing"
    RESPONSE_VALIDATION = "response_validation"
 
@dataclass
class PipelineConfig:
    """Centralized pipeline configuration"""
    # Model configuration
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 2048
 
    # Chain behavior
    max_retries: int = 3
    retry_backoff: float = 2.0
    timeout_seconds: int = 30
 
    # Memory management
    memory_type: str = "buffer"  # buffer, summary, entity
    max_memory_tokens: int = 4000
 
    # Context retrieval
    enable_rag: bool = True
    retrieval_strategy: str = "similarity"  # similarity, bm25, hybrid
    top_k_docs: int = 5
 
    # Logging and monitoring
    log_level: str = "INFO"
    enable_metrics: bool = True
    enable_tracing: bool = True
 
@dataclass
class StageContext:
    """Shared context across pipeline stages"""
    input_text: str
    metadata: Dict[str, Any]
    retrieved_docs: list = None
    prompt_template_used: str = None
    llm_response: str = None
    execution_time: Dict[str, float] = None
    errors: list = None

💡

**Configuration-First Design:** Store all pipeline parameters in centralized configs. This enables experimentation—swap temperature, memory types, or retrieval strategies without touching code.

Component Composition

Each stage is an independent, composable unit:

# pipelines/components/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict
import time
from datetime import datetime
 
class PipelineComponent(ABC):
    """Base class for all pipeline components"""
 
    def __init__(self, name: str, config: Dict[str, Any] = None):
        self.name = name
        self.config = config or {}
        self.execution_metrics = {
            "total_calls": 0,
            "successful_calls": 0,
            "failed_calls": 0,
            "total_time": 0.0,
            "avg_time": 0.0,
            "errors": []
        }
 
    @abstractmethod
    async def execute(self, context: StageContext) -> StageContext:
        """Execute component logic. Must be idempotent."""
        pass
 
    async def run(self, context: StageContext) -> StageContext:
        """Wrapper that handles timing, error tracking, metrics"""
        start_time = time.time()
 
        try:
            result = await self.execute(context)
            elapsed = time.time() - start_time
 
            # Update metrics
            self.execution_metrics["total_calls"] += 1
            self.execution_metrics["successful_calls"] += 1
            self.execution_metrics["total_time"] += elapsed
            self.execution_metrics["avg_time"] = (
                self.execution_metrics["total_time"] /
                self.execution_metrics["successful_calls"]
            )
 
            if context.execution_time is None:
                context.execution_time = {}
            context.execution_time[self.name] = elapsed
 
            return result
 
        except Exception as e:
            self.execution_metrics["failed_calls"] += 1
            self.execution_metrics["errors"].append({
                "timestamp": datetime.utcnow().isoformat(),
                "error": str(e),
                "component": self.name
            })
            raise
 
# pipelines/components/validators.py
import re
from typing import List, Optional
 
class InputValidator(PipelineComponent):
    """Validates input before pipeline execution"""
 
    def __init__(self, config: Dict[str, Any] = None):
        super().__init__("InputValidator", config)
        self.min_length = self.config.get("min_length", 10)
        self.max_length = self.config.get("max_length", 50000)
        self.forbidden_patterns = self.config.get("forbidden_patterns", [
            r"<script>",
            r"eval\(",
            r"__import__"
        ])
 
    async def execute(self, context: StageContext) -> StageContext:
        text = context.input_text.strip()
 
        # Length validation
        if len(text) < self.min_length:
            raise ValueError(f"Input too short: {len(text)} chars, minimum {self.min_length}")
        if len(text) > self.max_length:
            raise ValueError(f"Input too long: {len(text)} chars, maximum {self.max_length}")
 
        # Security checks
        for pattern in self.forbidden_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                raise ValueError(f"Forbidden pattern detected: {pattern}")
 
        # Normalize text
        context.input_text = self._normalize_text(text)
        context.metadata["validation_timestamp"] = datetime.utcnow().isoformat()
 
        return context
 
    def _normalize_text(self, text: str) -> str:
        """Normalize whitespace and encoding"""
        # Remove extra whitespace
        text = " ".join(text.split())
        # Normalize unicode
        return text.encode('utf-8', 'replace').decode('utf-8')
 
# pipelines/components/retrieval.py
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from typing import List
 
class ContextRetriever(PipelineComponent):
    """Retrieves relevant context using RAG"""
 
    def __init__(self, vector_store: Chroma, config: Dict[str, Any] = None):
        super().__init__("ContextRetriever", config)
        self.vector_store = vector_store
        self.top_k = self.config.get("top_k", 5)
        self.min_relevance_score = self.config.get("min_relevance_score", 0.5)
 
    async def execute(self, context: StageContext) -> StageContext:
        # Retrieve relevant documents
        docs_with_scores = self.vector_store.similarity_search_with_score(
            context.input_text,
            k=self.top_k * 2  # Over-retrieve to filter by relevance
        )
 
        # Filter by minimum relevance score
        relevant_docs = [
            doc for doc, score in docs_with_scores
            if (1 - score) >= self.min_relevance_score  # Convert distance to similarity
        ][:self.top_k]
 
        if not relevant_docs:
            context.retrieved_docs = []
            context.metadata["retrieval_status"] = "no_relevant_documents"
        else:
            context.retrieved_docs = relevant_docs
            context.metadata["retrieval_status"] = "success"
            context.metadata["retrieved_doc_count"] = len(relevant_docs)
 
        return context

⚠️

**Memory and Cost Implications:** Each RAG retrieval adds latency and costs. Use relevance thresholds aggressively, and implement caching for frequently retrieved contexts. Never over-retrieve.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦A permission model that splits agent authority by whether the side effect can be undone, not by how scary it feels

✦Measured before-and-after numbers for reprocessed items and API calls once checkpointing and caching were added

✦A complete resumable pipeline built on nothing but SQLite

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Advanced Memory Management

Memory Strategy Selection

LangChain's memory abstractions solve different problems. Choosing correctly determines whether your pipeline scales:

# pipelines/memory/memory_factory.py
from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationEntityMemory,
    ConversationBufferWindowMemory,
)
from enum import Enum
 
class MemoryStrategy(Enum):
    BUFFER = "buffer"              # All messages
    BUFFER_WINDOW = "buffer_window"  # Last N messages
    SUMMARY = "summary"            # Summarized history
    ENTITY = "entity"              # Entity tracking
 
class MemoryFactory:
    """Creates memory instances based on strategy and constraints"""
 
    @staticmethod
    def create_memory(
        strategy: MemoryStrategy,
        max_tokens: int = 4000,
        llm=None,
        **kwargs
    ):
        if strategy == MemoryStrategy.BUFFER:
            # Simple, low-latency, unbounded growth
            return ConversationBufferMemory(
                memory_key="chat_history",
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.BUFFER_WINDOW:
            # Fixed memory footprint, keeps recent context
            k = kwargs.get("k", 5)  # Number of messages to keep
            return ConversationBufferWindowMemory(
                memory_key="chat_history",
                k=k,
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.SUMMARY:
            # Compression via summarization (requires LLM)
            return ConversationSummaryMemory(
                llm=llm,
                memory_key="chat_history",
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.ENTITY:
            # Tracks entities and their attributes
            return ConversationEntityMemory(
                llm=llm,
                memory_key="chat_history",
                entity_extraction_prompt=kwargs.get("entity_extraction_prompt"),
                **kwargs
            )
 
# Usage with token limits
class TokenAwareMemory:
    """Manages memory with strict token budgets"""
 
    def __init__(self, max_tokens: int = 4000, llm=None):
        self.max_tokens = max_tokens
        self.llm = llm
        self._memory = None
        self._total_tokens = 0
        self._strategy = None
 
    def add_message(self, role: str, content: str):
        """Add message, switching strategy if needed"""
        # Estimate tokens (rough: 1 token ≈ 4 chars for English)
        estimated_tokens = len(content) // 4
 
        if self._total_tokens + estimated_tokens > self.max_tokens * 0.8:
            # Approaching limit, switch to summary strategy
            if self._strategy != MemoryStrategy.SUMMARY:
                self._switch_to_summary()
 
        self._memory.save_context(
            {"input": role},
            {"output": content}
        )
        self._total_tokens += estimated_tokens
 
    def _switch_to_summary(self):
        """Migrate from buffer to summary memory"""
        # Extract messages from current memory
        messages = self._memory.chat_memory.messages if hasattr(
            self._memory, 'chat_memory'
        ) else []
 
        # Create summary memory
        self._memory = MemoryFactory.create_memory(
            MemoryStrategy.SUMMARY,
            llm=self.llm,
            max_tokens=self.max_tokens
        )
        self._strategy = MemoryStrategy.SUMMARY
 
        # Re-add messages (triggers summarization)
        for msg in messages:
            self.add_message(msg.type, msg.content)

Distributed Session Memory

For multi-agent systems, memory must be persistent and queryable across sessions:

# pipelines/memory/persistent_memory.py
from typing import Dict, Any, List
import json
from datetime import datetime
import redis
import hashlib
 
class DistributedSessionMemory:
    """Persistent, distributed session memory backed by Redis"""
 
    def __init__(self, redis_client: redis.Redis, session_ttl: int = 86400):
        self.redis = redis_client
        self.session_ttl = session_ttl
 
    def save_session(self, session_id: str, messages: List[Dict[str, str]]):
        """Save session messages with TTL"""
        key = f"session:{session_id}:messages"
 
        serialized = json.dumps(messages, default=str)
        self.redis.setex(
            key,
            self.session_ttl,
            serialized
        )
 
    def get_session(self, session_id: str) -> List[Dict[str, str]]:
        """Retrieve session messages"""
        key = f"session:{session_id}:messages"
        data = self.redis.get(key)
        return json.loads(data) if data else []
 
    def query_sessions(self, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """Find all sessions for a user"""
        pattern = f"session:{user_id}:*:messages"
        keys = self.redis.keys(pattern)
 
        sessions = []
        for key in keys[:limit]:
            session_id = key.decode().split(':')[1]
            messages = self.get_session(session_id)
            if messages:
                sessions.append({
                    "session_id": session_id,
                    "message_count": len(messages),
                    "created_at": messages[0].get("timestamp"),
                    "last_updated": messages[-1].get("timestamp")
                })
 
        return sorted(sessions, key=lambda x: x["last_updated"], reverse=True)

Dynamic Routing and Chain Selection

Real-world pipelines don't execute the same chain for every input. Intelligent routing selects specialized chains based on input characteristics:

# pipelines/routing/router.py
from dataclasses import dataclass
from typing import Callable, List
from enum import Enum
 
@dataclass
class Route:
    """Represents a pipeline route"""
    name: str
    condition: Callable[[str], bool]
    chain: Any  # The LangChain chain to execute
    description: str
    priority: int = 0
 
class DynamicRouter(PipelineComponent):
    """Routes requests to specialized chains"""
 
    def __init__(self, routes: List[Route], config: Dict[str, Any] = None):
        super().__init__("DynamicRouter", config)
        # Sort by priority (higher first)
        self.routes = sorted(routes, key=lambda r: r.priority, reverse=True)
        self.fallback_chain = config.get("fallback_chain")
 
    async def execute(self, context: StageContext) -> StageContext:
        selected_route = None
 
        # Evaluate each route's condition
        for route in self.routes:
            try:
                if route.condition(context.input_text):
                    selected_route = route
                    break
            except Exception as e:
                context.metadata.setdefault("routing_errors", []).append({
                    "route": route.name,
                    "error": str(e)
                })
                continue
 
        # Use fallback if no route matched
        if not selected_route:
            if not self.fallback_chain:
                raise ValueError("No route matched and no fallback chain configured")
            selected_route = Route(
                name="fallback",
                condition=lambda x: True,
                chain=self.fallback_chain,
                description="Fallback chain"
            )
 
        context.metadata["selected_route"] = selected_route.name
        context.metadata["chain_description"] = selected_route.description
 
        return context
 
# Example usage
def create_document_qa_routes(config: PipelineConfig):
    """Create routes for document Q&A pipeline"""
 
    from langchain.chains import RetrievalQA
 
    # Simple retrieval for factual questions
    simple_qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=simple_retriever
    )
 
    # Complex reasoning for analytical questions
    reasoning_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="map_reduce",
        retriever=full_retriever
    )
 
    # Conversation for follow-ups
    conversation_chain = ConversationRetrievalChain.from_llm(
        llm=llm,
        retriever=conversational_retriever,
        memory=memory
    )
 
    return [
        Route(
            name="factual_qa",
            condition=lambda text: any(
                keyword in text.lower()
                for keyword in ["what is", "define", "explain", "who is"]
            ),
            chain=simple_qa_chain,
            description="Factual question answering",
            priority=10
        ),
        Route(
            name="analytical_qa",
            condition=lambda text: any(
                keyword in text.lower()
                for keyword in ["compare", "analyze", "why", "how does"]
            ),
            chain=reasoning_chain,
            description="Multi-document analysis",
            priority=9
        ),
        Route(
            name="conversation",
            condition=lambda text: len(memory.chat_memory.messages) > 0,
            chain=conversation_chain,
            description="Conversational follow-up",
            priority=8
        ),
    ]

Error Handling and Resilience

Production pipelines fail. The question is how gracefully:

# pipelines/resilience/retry_policy.py
import asyncio
from typing import Callable, Any, TypeVar
from enum import Enum
 
T = TypeVar('T')
 
class RetryStrategy(Enum):
    EXPONENTIAL = "exponential"
    LINEAR = "linear"
    FIXED = "fixed"
 
class ResilientComponent(PipelineComponent):
    """Base class for components with retry logic"""
 
    def __init__(
        self,
        name: str,
        config: Dict[str, Any] = None,
        max_retries: int = 3,
        strategy: RetryStrategy = RetryStrategy.EXPONENTIAL,
        base_delay: float = 1.0,
        max_delay: float = 30.0
    ):
        super().__init__(name, config)
        self.max_retries = max_retries
        self.strategy = strategy
        self.base_delay = base_delay
        self.max_delay = max_delay
 
    async def execute_with_retry(self, func: Callable[..., Any], *args, **kwargs):
        """Execute with automatic retry and backoff"""
        last_exception = None
 
        for attempt in range(self.max_retries + 1):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                last_exception = e
 
                if attempt == self.max_retries:
                    break
 
                # Calculate delay
                delay = self._calculate_delay(attempt)
 
                # Log retry
                self.execution_metrics["errors"].append({
                    "attempt": attempt + 1,
                    "error": str(e),
                    "retry_delay": delay
                })
 
                await asyncio.sleep(delay)
 
        raise last_exception
 
    def _calculate_delay(self, attempt: int) -> float:
        """Calculate delay based on retry strategy"""
        if self.strategy == RetryStrategy.EXPONENTIAL:
            delay = self.base_delay * (2 ** attempt)
        elif self.strategy == RetryStrategy.LINEAR:
            delay = self.base_delay * (attempt + 1)
        else:  # FIXED
            delay = self.base_delay
 
        return min(delay, self.max_delay)
 
class CircuitBreaker:
    """Prevents cascading failures from repeated retries"""
 
    class State(Enum):
        CLOSED = "closed"      # Normal operation
        OPEN = "open"          # Failing, reject requests
        HALF_OPEN = "half_open"  # Testing if recovered
 
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = self.State.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
 
    async def execute(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection"""
        if self.state == self.State.OPEN:
            if self._should_attempt_recovery():
                self.state = self.State.HALF_OPEN
                self.success_count = 0
            else:
                raise RuntimeError("Circuit breaker is OPEN")
 
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
 
    def _on_success(self):
        self.failure_count = 0
        if self.state == self.State.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = self.State.CLOSED
 
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = self.State.OPEN
 
    def _should_attempt_recovery(self) -> bool:
        if not self.last_failure_time:
            return True
        elapsed = time.time() - self.last_failure_time
        return elapsed >= self.recovery_timeout

Production Deployment and Monitoring

Pipeline Orchestration with Antigravity

Antigravity's orchestration layer handles deployment, scaling, and monitoring of complex pipelines:

# deploy/antigravity_pipeline.py
from antigravity import Agent, Pipeline, Config
import structlog
 
logger = structlog.get_logger()
 
class AIAssistantPipeline(Agent):
    """Production AI pipeline managed by Antigravity"""
 
    def __init__(self, config: PipelineConfig):
        super().__init__(name="ai-assistant-pipeline", version="1.0.0")
        self.config = config
 
        # Initialize components
        self.validator = InputValidator({"min_length": 10})
        self.retriever = ContextRetriever(vector_store, {"top_k": 5})
        self.router = DynamicRouter(routes, {"fallback_chain": default_chain})
        self.circuit_breaker = CircuitBreaker(failure_threshold=5)
 
        # Metrics tracking
        self.total_requests = 0
        self.successful_requests = 0
        self.failed_requests = 0
 
    async def execute(self, user_input: str, session_id: str) -> Dict[str, Any]:
        """Execute full pipeline"""
        self.total_requests += 1
 
        context = StageContext(
            input_text=user_input,
            metadata={"session_id": session_id, "timestamp": datetime.utcnow().isoformat()}
        )
 
        try:
            # Pipeline stages
            context = await self.validator.run(context)
            context = await self.retriever.run(context)
            context = await self.router.run(context)
 
            # Execute selected chain with circuit breaker
            response = await self.circuit_breaker.execute(
                self._execute_chain,
                context
            )
 
            self.successful_requests += 1
            return {
                "status": "success",
                "response": response,
                "metadata": context.metadata,
                "execution_time": sum(context.execution_time.values())
            }
 
        except Exception as e:
            self.failed_requests += 1
            logger.error(
                "pipeline_execution_failed",
                error=str(e),
                session_id=session_id,
                component=getattr(e, 'component', 'unknown')
            )
            return {
                "status": "error",
                "error": str(e),
                "metadata": context.metadata
            }
 
    async def _execute_chain(self, context: StageContext) -> str:
        """Execute the selected chain"""
        chain = context.metadata.get("chain")
        return await asyncio.to_thread(
            chain.run,
            context.input_text,
            callbacks=[self._create_callback_handler()]
        )
 
    def _create_callback_handler(self):
        """Create LangChain callbacks for logging and tracing"""
        from langchain.callbacks.base import BaseCallbackHandler
 
        class PipelineCallbackHandler(BaseCallbackHandler):
            def __init__(self, pipeline_agent):
                self.pipeline_agent = pipeline_agent
 
            def on_llm_start(self, serialized, prompts, **kwargs):
                logger.info("llm_call_started", prompt=prompts[0][:100])
 
            def on_llm_end(self, response, **kwargs):
                logger.info("llm_call_completed", token_usage=response.llm_output)
 
            def on_tool_start(self, serialized, input_str, **kwargs):
                logger.info("tool_execution", tool=serialized.get("name"), input=input_str)
 
        return PipelineCallbackHandler(self)
 
# Antigravity Configuration
pipeline_config = Config(
    name="AI Pipeline Service",
    description="Production-grade AI pipeline with LangChain integration",
    environment="production",
    max_concurrency=100,
    timeout_seconds=30,
    enable_tracing=True,
    enable_metrics=True,
    alert_thresholds={
        "error_rate": 0.05,  # Alert if >5% errors
        "p95_latency": 15000,  # Alert if p95 > 15s
        "circuit_breaker_open": True
    }
)
 
pipeline_agent = AIAssistantPipeline(config)

Monitoring and Observability

# monitoring/pipeline_metrics.py
from prometheus_client import Counter, Histogram, Gauge
import time
 
class PipelineMetrics:
    """Prometheus metrics for pipeline monitoring"""
 
    def __init__(self):
        self.request_counter = Counter(
            'pipeline_requests_total',
            'Total pipeline requests',
            ['status', 'route']
        )
 
        self.latency_histogram = Histogram(
            'pipeline_latency_seconds',
            'Pipeline execution latency',
            ['stage'],
            buckets=(0.1, 0.5, 1, 2, 5, 10)
        )
 
        self.token_usage_gauge = Gauge(
            'llm_tokens_used',
            'Tokens used by LLM',
            ['direction']  # 'input' or 'output'
        )
 
        self.memory_size_gauge = Gauge(
            'memory_size_tokens',
            'Current memory size in tokens',
            ['session_id']
        )
 
    def record_request(self, status: str, route: str, duration: float):
        self.request_counter.labels(status=status, route=route).inc()
        self.latency_histogram.labels(stage=route).observe(duration)
 
    def record_token_usage(self, input_tokens: int, output_tokens: int):
        self.token_usage_gauge.labels(direction='input').inc(input_tokens)
        self.token_usage_gauge.labels(direction='output').inc(output_tokens)
 
# Export metrics endpoint
from flask import Flask, Response
from prometheus_client import generate_latest
 
metrics_app = Flask(__name__)
metrics = PipelineMetrics()
 
@metrics_app.route('/metrics')
def prometheus_metrics():
    return Response(generate_latest(), mimetype='text/plain')

Testing Pipelines at Scale

Production pipelines require comprehensive testing strategies:

# tests/test_pipeline.py
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock
 
@pytest.mark.asyncio
async def test_pipeline_happy_path():
    """Test successful pipeline execution"""
    config = PipelineConfig()
    pipeline = AIAssistantPipeline(config)
 
    result = await pipeline.execute(
        "What is machine learning?",
        session_id="test-session-1"
    )
 
    assert result["status"] == "success"
    assert "response" in result
    assert result["metadata"]["validation_timestamp"]
 
@pytest.mark.asyncio
async def test_pipeline_validation_failure():
    """Test handling of invalid input"""
    pipeline = AIAssistantPipeline(PipelineConfig())
 
    result = await pipeline.execute(
        "x",  # Too short
        session_id="test-session-2"
    )
 
    assert result["status"] == "error"
    assert "too short" in result["error"].lower()
 
@pytest.mark.asyncio
async def test_circuit_breaker_opens():
    """Test circuit breaker behavior under failures"""
    breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=1)
 
    # First 3 failures
    for i in range(3):
        with pytest.raises(RuntimeError):
            await breaker.execute(AsyncMock(side_effect=RuntimeError("Failure")))
 
    # Circuit should be open
    assert breaker.state == CircuitBreaker.State.OPEN
 
    with pytest.raises(RuntimeError, match="Circuit breaker is OPEN"):
        await breaker.execute(AsyncMock())
 
@pytest.mark.asyncio
@pytest.mark.benchmark
async def test_pipeline_load(benchmark):
    """Benchmark pipeline under load"""
    pipeline = AIAssistantPipeline(PipelineConfig())
 
    async def run_pipeline():
        return await pipeline.execute("Test query", "session-load-test")
 
    # Simulate 100 concurrent requests
    await asyncio.gather(*[run_pipeline() for _ in range(100)])
 
    # Check performance
    assert pipeline.total_requests == 100
    assert pipeline.failed_requests < 5  # Allow <5% failures

How Much Execution Authority Should an Agent Get

Once a pipeline starts acting on its own, the center of gravity shifts from how do these steps connect to what is this thing allowed to do.

Antigravity v2.2.1 (released 2026-06-25) introduced a unified permission model that lets you declare an agent's operating scope explicitly. I treat it as more than a safety feature — it's the vocabulary for pipeline design. And I draw the tiers by reversibility of the side effect, not by category of function.

Tier	Example operations	Reversible?	Approval
L1 Observe	Search, file reads, metrics fetch	Fully (no side effect)	None
L2 Reversible write	Commit to a branch, cache update, save a draft	Can be undone	Review after the fact
L3 Irreversible	Production deploy, charging a card, sending mail, DB DELETE	Cannot be undone	Human approval first

Draw the line at "is this dangerous" and every new tool becomes a fresh judgment call. Draw it at reversibility and the answer is determined the moment you write the function.

Here is how I enforce it at the LangChain tool layer. Every tool declares its tier, and an L3 tool cannot run without passing through an approval callback.

# pipeline/permissions.py
from enum import IntEnum
from dataclasses import dataclass
from typing import Callable, Optional, Any
import logging
 
logger = logging.getLogger(__name__)
 
 
class Tier(IntEnum):
    OBSERVE = 1        # no side effects
    REVERSIBLE = 2     # writes that can be undone
    IRREVERSIBLE = 3   # writes that cannot
 
 
class PermissionDenied(RuntimeError):
    """The operation was not approved."""
 
 
@dataclass
class PermissionPolicy:
    """Highest tier allowed to run without a human in the loop."""
    auto_max: Tier = Tier.REVERSIBLE
    approve: Optional[Callable[[str, dict], bool]] = None
 
    def check(self, tool_name: str, tier: Tier, payload: dict) -> None:
        if tier <= self.auto_max:
            logger.info("auto-approved", extra={"tool": tool_name, "tier": int(tier)})
            return
        if self.approve is None:
            raise PermissionDenied(f"{tool_name}: no approval handler for L{int(tier)}")
        if not self.approve(tool_name, payload):
            raise PermissionDenied(f"{tool_name}: human approval was not granted")
        logger.warning("human-approved", extra={"tool": tool_name, "tier": int(tier)})
 
 
def guarded(tier: Tier, policy: PermissionPolicy):
    """Bind a permission tier to a tool function."""
    def wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
        def inner(**kwargs: Any) -> Any:
            policy.check(fn.__name__, tier, kwargs)
            return fn(**kwargs)
        inner.__name__ = fn.__name__
        inner.__doc__ = fn.__doc__
        inner.permission_tier = tier  # type: ignore[attr-defined]
        return inner
    return wrap

On the calling side, forgetting to declare a tier is not an option — an undeclared function never becomes a tool.

# pipeline/tools.py
from langchain_core.tools import StructuredTool
from pipeline.permissions import Tier, PermissionPolicy, guarded
 
policy = PermissionPolicy(
    auto_max=Tier.REVERSIBLE,
    approve=lambda name, payload: input(f"Run {name}? {payload} [y/N] ") == "y",
)
 
 
@guarded(Tier.OBSERVE, policy)
def search_docs(query: str) -> str:
    """Search the docs and return the matching passage."""
    return retriever.invoke(query)[0].page_content
 
 
@guarded(Tier.IRREVERSIBLE, policy)
def publish_release(tag: str) -> str:
    """Publish a release to production."""
    return deploy_client.publish(tag)
 
 
def as_tools(*fns) -> list[StructuredTool]:
    for fn in fns:
        if not hasattr(fn, "permission_tier"):
            raise ValueError(f"{fn.__name__} has no declared permission tier")
    return [StructuredTool.from_function(fn) for fn in fns]

Why a decorator rather than a prompt? I tried the prompt route — "please confirm before anything destructive." That is an instruction, not a constraint. Constraints belong in code, or they evaporate the day the model changes underneath you.

I look at the same boundary from a different angle in Three Boundaries I Draw Before Handing Work to an Antigravity 2.0 Agent.

Measure the Failures and the Cost Before You Choose an Architecture

Back to that overnight run. What it lacked wasn't retries. It lacked a checkpoint.

A retry repeats one call. It has no memory of how far the run got.

Checkpointing needs no infrastructure worth the name. SQLite is enough.

# pipeline/checkpoint.py
import sqlite3
import json
from typing import Iterable, Iterator, Any
 
SCHEMA = """
CREATE TABLE IF NOT EXISTS progress (
    run_id TEXT NOT NULL,
    item_key TEXT NOT NULL,
    result TEXT,
    PRIMARY KEY (run_id, item_key)
);
"""
 
 
class Checkpoint:
    def __init__(self, path: str = "pipeline_state.db") -> None:
        self.conn = sqlite3.connect(path, isolation_level=None)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.executescript(SCHEMA)
 
    def done_keys(self, run_id: str) -> set[str]:
        cur = self.conn.execute(
            "SELECT item_key FROM progress WHERE run_id = ?", (run_id,)
        )
        return {row[0] for row in cur}
 
    def record(self, run_id: str, item_key: str, result: Any) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO progress VALUES (?, ?, ?)",
            (run_id, item_key, json.dumps(result, ensure_ascii=False)),
        )
 
 
def resumable(
    run_id: str,
    items: Iterable[tuple[str, Any]],
    ckpt: Checkpoint,
) -> Iterator[tuple[str, Any]]:
    """Yield only the items this run hasn't finished yet."""
    already = ckpt.done_keys(run_id)
    for key, payload in items:
        if key in already:
            continue
        yield key, payload

The caller does nothing clever. It records each result as it lands.

# pipeline/run.py
from pipeline.checkpoint import Checkpoint, resumable
 
ckpt = Checkpoint()
RUN_ID = "wallpaper-classify-2026-03"
 
for key, image in resumable(RUN_ID, iter_assets(), ckpt):
    try:
        result = classification_chain.invoke({"image": image})
    except RateLimitError:
        # Whatever was recorded is skipped on the next run
        raise
    ckpt.record(RUN_ID, key, result)

INSERT OR REPLACE keeps a replayed key from corrupting anything. Mix in a non-idempotent write and resumption becomes the source of the next outage.

Here is what I measured classifying a few thousand assets on my own machine. Same input set, same model, same rate limit, three configurations — each killed once mid-run and then restarted.

Configuration	Items reprocessed after the kill	API calls on resume	Extra time to finish
Retries only	3,200 (everything)	3,200	~48 min
+ Checkpointing	~180	180	~3 min
+ Embedding cache	~180	62	~1 min

The absolute numbers travel poorly between environments. The ratios are the point. Checkpointing turns the cost of resuming into roughly a constant; caching then shaves what remains. Add the cache first and a crash still costs you the whole night. Checkpointing pays first.

I learned that ordering by watching my cost estimates miss. Raising cache hit rates is satisfying work with visible progress. But as long as the nightly batch dies and reruns from zero, none of that satisfaction shows up on the invoice.

For how failures get classified and recovered a layer down, AI Agent Error Recovery Design: Building Pipelines That Don't Stop pairs well with this. And if concurrency itself is your bottleneck, Stop Hard-Coding Your Agent Concurrency covers tuning it from observed backpressure.

The One Line to Add Tomorrow

Talk about pipeline design as "choosing an architecture" and the diagrams arrive first. What actually moved the needle for me was plainer than that.

Where does this resume when it dies. Can that operation be undone. What did that single item cost.

Get to a state where you can answer those three, and the routing and memory optimizations that follow stop generating rework. Done in the other order — as I did — you end up with a beautifully composed chain that you rerun from scratch every morning.

So here's the concrete next step: drop a Checkpoint into the pipeline you're running today, then kill it on purpose halfway through. If it resumes in minutes, everything you optimize afterward compounds. If it takes tens of minutes, the chain composition was never the thing to fix.

Thank you for reading — I hope some of this proves useful in your own runs.

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →

◈ Agents & Manager2026-04-27

Designing a Context Compression Sub-Agent in AgentKit 2.0 — A Production Pipeline That Stops Long-Running Tasks From Drowning in Their Own History

When you run an AgentKit 2.0 agent for hours inside Antigravity, response time and accuracy degrade together as the conversation history balloons. This premium guide walks through a production-grade dynamic compression sub-agent — schema, prompt, integration hooks, and an A/B evaluation harness — so you can decide rollout based on numbers, not gut feel.

◈ Agents & Manager2026-04-23

Production Multi-Agent Systems with Antigravity AgentKit 2.0: Patterns, Failure Modes, and What the Demos Don't Show

AgentKit 2.0 makes multi-agent systems look effortless in demos, but running them in production is a different problem. This guide covers Planning vs. Fast mode, three real orchestration patterns, and the failure modes — infinite loops, cost blowouts, prompt injection — that bite on day one.

◈ Agents & Manager2026-04-07

Antigravity × LangGraph: Complete Masterguide for Stateful AI Agent Development

A comprehensive guide to building stateful AI agents by combining LangGraph with Antigravity. Covers graph-based workflow design, state management, checkpointing, and human-in-the-loop patterns for production-ready systems.

📚RECOMMENDED BOOKS

Build a Large Language Model (From Scratch)

Sebastian Raschka

LLM Dev

Prompt Engineering for LLMs

* Contains affiliate links

See all →

LangChain and Antigravity Pipelines, Designed Backwards from Permissions, Failures, and Cost

Understanding AI Pipeline Architecture

The Pipeline Abstraction

Five Pipeline Patterns

Building Sequential Pipelines with Composable Components

Core Architecture

Component Composition

Continue Reading

Unlock This Article

Advanced Memory Management

Memory Strategy Selection

Distributed Session Memory

Dynamic Routing and Chain Selection

Error Handling and Resilience

Production Deployment and Monitoring

Pipeline Orchestration with Antigravity

Monitoring and Observability

Testing Pipelines at Scale

How Much Execution Authority Should an Agent Get

Measure the Failures and the Cost Before You Choose an Architecture

The One Line to Add Tomorrow

Thank You for Reading

Related Articles