AgentKit (LangChain Agent Runtime)

LangChain's production agent runtime: graph-based, streaming, checkpointing, horizontal scaling.

Overview

AgentKit is LangChain's agent runtime — the execution layer for LangGraph graphs in production. It provides:

• Horizontal scaling — Multiple workers, shared state
• Streaming — Token-level, structured, custom
• Checkpointing — Postgres, SQLite, Redis
• Multi-tenancy — Thread isolation, auth
• Observability — LangSmith native
• API — REST, WebSocket, GraphQL

Architecture

┌─────────────────────────────────────┐
│           AgentKit Server           │
├─────────────────────────────────────┤
│  API Layer (REST/WS/GraphQL)        │
├─────────────────────────────────────┤
│  Execution Engine                   │
│  ├── Graph Executor                 │
│  ├── Stream Manager                 │
│  ├── Checkpoint Coordinator         │
│  └── Tenant Manager                 │
├─────────────────────────────────────┤
│  State Backend (Postgres/Redis)     │
├─────────────────────────────────────┤
│  LangSmith (Observability)          │
└─────────────────────────────────────┘

Quick Start

1. Define Graph (LangGraph)

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

class State(TypedDict):
    messages: Annotated[list, operator.add]
    user_id: str

def reasoning(state):
    # Your logic
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("reasoning", reasoning)
graph.add_edge(START, "reasoning")
graph.add_edge("reasoning", END)

compiled = graph.compile()

2. Configure AgentKit

# agentkit.yaml
server:
  host: "0.0.0.0"
  port: 8000
  workers: 4

database:
  url: "postgresql://user:pass@localhost/agentkit"
  pool_size: 20

redis:
  url: "redis://localhost:6379/0"

langsmith:
  api_key: "${LANGSMITH_API_KEY}"
  project: "my-agents"

auth:
  enabled: true
  jwt_secret: "${JWT_SECRET}"
  allowed_origins: ["https://app.example.com"]

3. Run

# Install
pip install agentkit

# Run server
agentkit serve --config agentkit.yaml

# Or programmatically
from agentkit import AgentKitServer

server = AgentKitServer.from_config("agentkit.yaml")
server.add_graph("my_agent", compiled)
server.run()

API Reference

REST Endpoints

Create Thread

curl -X POST http://localhost:8000/threads \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"graph_id": "my_agent", "metadata": {"user_id": "123"}}'

Run Graph

curl -X POST http://localhost:8000/threads/$THREAD_ID/runs \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"input": {"messages": [{"role": "user", "content": "Hello"}]}}'

Stream

curl -N http://localhost:8000/threads/$THREAD_ID/runs/$RUN_ID/stream \
  -H "Authorization: Bearer $TOKEN"
# SSE stream: data: {"type": "token", "data": "..."}

Multi-Tenancy

Thread Isolation

# Each thread = isolated conversation
thread_config = {
    "configurable": {
        "thread_id": "tenant-123:session-456",
        "user_id": "user-789"
    }
}

# State scoped to thread
state = await server.get_thread_state(thread_id)

Authentication

# JWT-based
# Header: Authorization: Bearer <jwt>
# JWT claims: { "sub": "user-123", "tenant": "acme-corp", "roles": ["user"] }

# API Key
# Header: X-API-Key: ak_...

Tenant Configuration

# agentkit.yaml
tenants:
  acme-corp:
    rate_limit: 1000  # req/min
    max_threads: 10000
    max_run_time: 300  # seconds
    allowed_graphs: ["support_agent", "sales_agent"]
    pii_masking: true

Checkpointing

Backends

Configuration

checkpointer:
  type: "postgres"
  url: "postgresql://user:pass@localhost/agentkit"
  pool_size: 20
  # Table: checkpoint_writes, checkpoint_blobs
  # Auto-migrates schema

Migration

# Schema versioning automatic
# Custom migration
from agentkit.migrations import migrate

migrate(
    from_version=1,
    to_version=2,
    migration_fn=lambda state: {
        **state,
        "schema_version": 2,
        "new_field": "default"
    }
)

Streaming

Stream Modes

# In API request
{
    "input": {...},
    "stream_mode": "messages"  # or "values", "updates", "debug"
}

Server-Sent Events (SSE)

// Client
const evtSource = new EventSource(`/threads/${threadId}/runs/${runId}/stream`, {
  headers: { Authorization: `Bearer ${token}` }
});

evtSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  if (data.type === "token") {
    appendToUI(data.data);
  }
};

Custom Streaming

# In graph node
def streaming_node(state):
    for token in llm.stream(state["prompt"]):
        yield {"custom_event": token}  # Custom stream mode

Scaling

Horizontal Scaling

# agentkit.yaml
server:
  workers: 4  # Processes
  threads_per_worker: 100

# Behind load balancer
# Sticky sessions NOT required (stateless workers)
# State in Postgres/Redis

Queue Management

queue:
  max_size: 10000
  timeout: 300  # Max wait time
  priority_levels: 3
  dead_letter_ttl: 86400

Rate Limiting

rate_limit:
  global: 10000  # req/min
  per_tenant: 1000
  per_user: 100
  burst: 50

Observability

LangSmith Integration

langsmith:
  api_key: "${LANGSMITH_API_KEY}"
  project: "production-agents"
  tags: ["production", "v2.1"]
  metadata:
    environment: "production"
    region: "us-east-1"

Custom Metrics

from agentkit.metrics import MetricsCollector

metrics = MetricsCollector()

@metrics.track("agent_runtime")
async def my_node(state):
    # Automatically tracks: latency, tokens, errors
    return result

Health Checks

# Liveness
curl http://localhost:8000/health/live

# Readiness
curl http://localhost:8000/health/ready
# Checks: DB connection, Redis, graph compilation

Deployment

Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY agent.py agentkit.yaml .
CMD ["agentkit", "serve", "--config", "agentkit.yaml"]

Kubernetes

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agentkit
spec:
  replicas: 4
  selector:
    matchLabels:
      app: agentkit
  template:
    spec:
      containers:
      - name: agentkit
        image: myorg/agentkit:latest
        ports:
        - containerPort: 8000
        envFrom:
        - secretRef:
            name: agentkit-secrets
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
---
# Service
apiVersion: v1
kind: Service
metadata:
  name: agentkit
spec:
  selector:
    app: agentkit
  ports:
  - port: 80
    targetPort: 8000

LangGraph Platform (Managed)

# Zero-config managed AgentKit
langgraph deploy --project my-agents
# Auto-scaling, managed Postgres, LangSmith, auth

Migration from LangGraph Direct

Related Notes

• LangGraph — Graph definition
• Framework Comparison — vs others
• Checkpointing — Checkpointing patterns
• Observability — LangSmith
• Deployment Architecture — K8s, scaling
• Memory Systems — Thread-scoped memory
• Security — Multi-tenancy, auth

Sources: AgentKit docs (langchain-ai.github.io/agentkit), LangGraph Platform docs, LangChain blog, AgentKit GitHub, production deployment guides.