Tamiz Uddin

Posted on Oct 26 Originally published at tamiz.pro

Why Your AI Agent Stack Needs a Gateway, Not a Glue Layer: Lessons from the 11k-Star Orchestration War

#ai #machine #learning #agents

Originally published on tamiz.pro.

The AI agent landscape has shifted rapidly from simple prompt-engineering experiments to complex, multi-step autonomous systems. As these systems grow in sophistication, the underlying architecture that connects LLMs, tools, memory, and workflows becomes the critical differentiator between a fragile prototype and a production-grade platform. The recent

The recent explosion of open-source orchestration frameworks—LangChain, LlamaIndex, AutoGen, CrewAI, and the like—has given developers a false sense of security. They ship a "glue layer": a collection of prompt templates, chain abstractions, and tool wrappers that make the first demo feel magical. But when latency spikes, token costs balloon, or a downstream API deprecates a parameter, the glue dissolves. You're left patching Runnable subclasses at 3 a.m., wondering why your "agent" just sent 40 k tokens to summarize a two-sentence email.

A gateway is not a thicker glue layer. It is a distinct infrastructure tier that sits between your orchestration logic and every external dependency—LLM providers, vector stores, tool APIs, auth systems, observability backends. It enforces contracts, centralizes cross-cutting concerns, and gives you a single lever to pull when the inevitable "provider X changed their response format" incident hits.

1. Glue Layer vs. Gateway: The Architectural Divide

Concern	Glue Layer (Library)	Gateway (Service)
Deployment	Embedded in your process	Independent, horizontally scalable
Failure domain	Crashes your app	Isolated; circuit-breakers protect callers
Observability	Scattered logs/metrics	Unified request IDs, distributed traces, cost dashboards
Policy enforcement	Ad-hoc `if` statements in chains	Declarative: rate limits, PII redaction, model routing
Upgrade path	Rebuild & redeploy every consumer	Rolling deploy; consumers unchanged
Multi-tenancy	Nearly impossible	Native: per-tenant quotas, model allow-lists, data residency

The 11k-star project that taught us this lesson the hard way is LangChain. Its early LLMChain and AgentExecutor classes were pure glue. Teams that treated them as infrastructure—baking provider-specific logic into custom chains—spent quarters migrating to LangGraph or LangServe when streaming, human-in-the-loop, and multi-agent handoffs became requirements. The teams that wrapped LangChain behind a thin gateway service from day one swapped providers, added guardrails, and enabled multi-tenancy without touching their business logic.

2. Minimal Gateway Contract

A gateway needs only three stable interfaces. Everything else is implementation detail.

# gateway/contracts.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Literal

@dataclass(frozen=True)
class ChatMessage:
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    tool_calls: list[dict] | None = None
    tool_call_id: str | None = None

@dataclass(frozen=True)
class ChatRequest:
    messages: list[ChatMessage]
    model: str | None = None          # None → gateway picks default
    temperature: float = 0.7
    max_tokens: int | None = None
    stream: bool = False
    metadata: dict | None = None      # tenant_id, trace_id, cost_center…

@dataclass(frozen=True)
class ChatResponse:
    message: ChatMessage
    usage: dict                       # prompt_tokens, completion_tokens, cost_usd
    model: str                        # actual model used (after routing)
    latency_ms: int

class Gateway(ABC):
    @abstractmethod
    async def chat(self, req: ChatRequest) -> ChatResponse: ...

    @abstractmethod
    async def chat_stream(self, req: ChatRequest) -> AsyncIterator[ChatResponse]: ...

    @abstractmethod
    async def health(self) -> bool: ...

That's it. Your orchestration code (LangGraph, custom state machines, whatever) depends only on this ABC. The gateway implementation handles:

Provider routing (OpenAI ↔ Anthropic ↔ local vLLM ↔ Bedrock)
Retry/timeout/circuit-breaker policies
Token counting & cost accounting
PII scrubbing / structured output validation
Request/response logging to ClickHouse / Datadog / OTel

3. Reference Implementation: 120 Lines of Production-Grade Gateway

# gateway/impl.py
import os, time, uuid, json, logging
from contextlib import asynccontextmanager
from dataclasses import asdict
from typing import AsyncIterator

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

from .contracts import ChatMessage, ChatRequest, ChatResponse, Gateway

log = logging.getLogger("gateway")

PROVIDER_CONFIG = {
    "openai": {
        "base_url": "https://api.openai.com/v1",
        "models": ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"],
        "auth": lambda: {"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
    },
    "anthropic": {
        "base_url": "https://api.anthropic.com/v1",
        "models": ["claude-3-5-sonnet-20241022", "claude-3-haiku-20240307"],
        "auth": lambda: {"x-api-key": os.getenv("ANTHROPIC_API_KEY")},
    },
}

class HttpGateway(Gateway):
    def __init__(self, default_model: str = "gpt-4o-mini"):
        self.default_model = default_model
        self._client = httpx.AsyncClient(timeout=60.0, limits=httpx.Limits(max_connections=100))

    def _pick_provider(self, model: str | None) -> tuple[str, dict]:
        model = model or self.default_model
        for name, cfg in PROVIDER_CONFIG.items():
            if model in cfg["models"]:
                return name, cfg
        raise ValueError(f"No provider configured for model {model}")

    @retry(wait=wait_exponential_jitter(initial=1, max=10), stop=stop_after_attempt(3))
    async def _post(self, provider: str, path: str, payload: dict, headers: dict, stream: bool):
        cfg = PROVIDER_CONFIG[provider]
        url = f"{cfg['base_url']}{path}"
        req_headers = {**cfg["auth"](), **headers, "Content-Type": "application/json"}
        if stream:
            return self._client.stream("POST", url, json=payload, headers=req_headers)
        resp = await self._client.post(url, json=payload, headers=req_headers)
        resp.raise_for_status()
        return resp.json()

    def _to_provider_messages(self, msgs: list[ChatMessage]) -> list[dict]:
        out = []
        for m in msgs:
            d = {"role": m.role, "content": m.content}
            if m.tool_calls:
                d["tool_calls"] = m.tool_calls
            if m.tool_call_id:
                d["tool_call_id"] = m.tool_call_id
            out.append(d)
        return out

    async def chat(self, req: ChatRequest) -> ChatResponse:
        provider, cfg = self._pick_provider(req.model)
        start = time.perf_counter()
        payload = {
            "model": req.model or self.default_model,
            "messages": self._to_provider_messages(req.messages),
            "temperature": req.temperature,
            "stream": False,
        }
        if req.max_tokens:
            payload["max_tokens"] = req.max_tokens

        data = await self._post(provider, "/chat/completions", payload, {}, False)
        latency = int((time.perf_counter() - start) * 1000)

        choice = data["choices"][0]
        msg = ChatMessage(role=choice["message"]["role"], content=choice["message"].get("content", ""))
        usage = data.get("usage", {})
        return ChatResponse(
            message=msg,
            usage={**usage, "cost_usd": self._estimate_cost(provider, req.model or self.default_model, usage)},
            model=data["model"],
            latency_ms=latency,
        )

    async def chat_stream(self, req: ChatRequest) -> AsyncIterator[ChatResponse]:
        provider, cfg = self._pick_provider(req.model)
        payload = {
            "model": req.model or self.default_model,
            "messages": self._to_provider_messages(req.messages),
            "temperature": req.temperature,
            "stream": True,
        }
        if req.max_tokens:
            payload["max_tokens"] = req.max_tokens

        async with await self._post(provider, "/chat/completions", payload, {}, True) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                if line.strip() == "data: [DONE]":
                    break
                chunk = json.loads(line[6:])
                delta = chunk["choices"][0].get("delta", {})
                if "content" in delta:
                    yield ChatResponse(
                        message=ChatMessage(role="assistant", content=delta["content"]),
                        usage={},  # usage only in final chunk
                        model=chunk.get("model", req.model or self.default_model),
                        latency_ms=0,
                    )

    def _estimate_cost(self, provider: str, model: str, usage: dict) -> float:
        # Simplified; replace with real pricing table
        rates = {"gpt-4o": 0.005, "gpt-4o-mini": 0.00015, "claude-3-5-sonnet-20241022": 0.003}
        in_rate = rates.get(model, 0.001)
        out_rate = in_rate * 3
        return (usage.get("prompt_tokens", 0) * in_rate + usage.get("completion_tokens", 0) * out_rate) / 1000

    async def health(self) -> bool:
        try:
            await self._client.get("https://api.openai.com/v1/models", timeout=5.0)
            return True
        except Exception:
            return False

    async def aclose(self):
        await self._client.aclose()

Key properties:

Zero business logic — no prompt templates, no agent loops, no memory.
Provider-agnostic — adding Bedrock or a local vLLM endpoint is a 5-line config change.
Observable by default — every call returns usage, latency_ms, model; wrap with OTel middleware for traces.
Resilient — tenacity retries with jitter, connection pooling, circuit-breaker-ready (add pybreaker if needed).
Testable — swap HttpGateway for a FakeGateway in unit tests; no mocking httpx.

4. Wiring It Into Your Orchestration (LangGraph Example)

# orchestrator/agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

from gateway.contracts import ChatRequest, ChatMessage, Gateway
from gateway.impl import HttpGateway

class AgentState(TypedDict):
    messages: Annotated[list[ChatMessage], operator.add]
    gateway: Gateway  # injected at runtime

async def call_model(state: AgentState) -> AgentState:
    gw: Gateway = state["gateway"]
    req = ChatRequest(messages=state["messages"], model="gpt-4o-mini", temperature=0.2)
    resp = await gw.chat(req)
    return {"messages": [resp.message]}

graph = StateGraph(AgentState)
graph.add_node("llm", call_model)
graph.set_entry_point("llm")
graph.add_edge("llm", END)
app = graph.compile()

# Runtime bootstrap
async def run(user_input: str):
    gw = HttpGateway()
    try:
        init_state = {"messages": [ChatMessage(role="user", content=user_input)], "gateway": gw}
        async for event in app.astream(init_state):
            print(event)
    finally:
        await gw.aclose()

Notice: the graph knows nothing about OpenAI, Anthropic, retries, or cost. It only knows Gateway. Swap HttpGateway for a LocalLLMGateway (vLLM, Ollama) in staging; prod stays unchanged.

5. Cross-Cutting Concerns You Get For Free

Concern	Glue-Layer Approach	Gateway Approach
PII redaction	Scatter `presidio` calls in every chain	One middleware in gateway: `req.messages = redact(req.messages)`
Structured output enforcement	`PydanticOutputParser` per chain	Gateway validates against JSON Schema before returning to caller
Cost guardrails	Manual `if usage > budget: raise`	Central policy: `max_cost_usd_per_request`, `daily_quota_per_tenant`
Model deprecation	Find/replace model strings across 50 files	Update `PROVIDER_CONFIG`; gateway routes `gpt-3.5-turbo` → `gpt-4o-mini` automatically
Multi-region latency	Hardcode `base_url` per env	Gateway reads `X-Region` header, routes to nearest endpoint
Audit log	`logging.info` in 20 places	Single `log.info(json.dumps({...}))` in `chat()` — immutable, queryable

6. When Not to Build a Gateway

Throwaway prototypes (< 2 weeks, single developer, one provider).
Purely local inference (Ollama + no external tools) where a thin wrapper suffices.
Serverless functions with < 100 ms budget — the extra network hop hurts. (Solution: co-locate gateway as a sidecar.)

If you're past the prototype phase, the gateway pays for itself the first time a provider drops a parameter or you need to onboard a second LLM vendor.

7. Migration Path for Existing Glue-Layer Codebases

Extract all direct openai.AsyncClient() / anthropic.AsyncClient() calls into a single module.
Implement the Gateway ABC over that module (zero behavior change).
Inject the gateway instance into your orchestration entry points (FastAPI Depends, LangGraph state, etc.).
Enable observability: add request IDs, log usage, hook up Grafana dashboard.
Iterate: add routing, guardrails, multi-tenancy one policy at a time without touching orchestration code.

8. Closing Thought

The orchestration war isn't won by the prettiest chain syntax or the most agents running in parallel. It's won by the team that can swap the model, add a guardrail, or onboard a new tenant on a Friday afternoon without paging the on-call. A gateway gives you that lever. A glue layer gives you a blog post.

Build the gateway. Keep the glue in the application layer where it belongs—thin, testable, and replaceable.

DEV Community