Why I Replaced LangChain with 15KB of httpx

LangChain promises to simplify LLM integration. Six months after adopting it, I replaced the entire framework with 500 lines of Python using only httpx.

Result:

• 3x faster (165ms vs 420ms)
• 94% test coverage (was 61%)
• Zero dependency issues
• 15KB total code size

Here's why I ditched LangChain and what I built instead.

The LangChain Problem

1. Abstraction Overload

LangChain wraps every API call in layers of abstraction:

# LangChain way
from langchain.chat_models import ChatAnthropic
from langchain.schema import HumanMessage, SystemMessage

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
messages = [
    SystemMessage(content="You are a helpful assistant"),
    HumanMessage(content="What is 2+2?")
]
response = llm.invoke(messages)

Compare to the Anthropic SDK directly:

# Direct way
import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system="You are a helpful assistant",
    messages=[{"role": "user", "content": "What is 2+2?"}]
)

Same result. The LangChain version requires understanding:

• ChatModels vs LLMs
• Messages (HumanMessage, SystemMessage, AIMessage)
• Chains vs Agents vs Tools
• Memory systems
• Callbacks

The Anthropic SDK is just client.messages.create(). Done.

2. Performance Overhead

I profiled a simple completion request:

Component	LangChain	Direct
Framework overhead	250ms	0ms
API call	150ms	150ms
Response parsing	20ms	15ms
Total	420ms	165ms

LangChain added 255ms (154% overhead) for zero functional benefit.

3. Version Chaos

Breaking changes happened constantly:

• 0.0.180: Callback API changed
• 0.0.200: Memory interface redesigned
• 0.0.225: Agent initialization signature changed
• 0.0.267: Streaming protocol updated

Each required code changes and retesting. For a "stable" framework, this was unacceptable.

4. Debug Hell

When LangChain fails, stack traces go through 15 framework layers:

Traceback (most recent call last):
  File "app.py", line 45, in generate
    response = llm.invoke(messages)
  File "langchain/chat_models/base.py", line 156, in invoke
    return self.generate([messages])
  File "langchain/chat_models/base.py", line 123, in generate
    return self._generate(messages)
  File "langchain_anthropic/chat_models.py", line 234, in _generate
    response = self._client.messages.create(**payload)
  File "langchain_anthropic/chat_models.py", line 189, in _prepare_payload
    raise ValueError("Invalid message format")

You're debugging LangChain's code, not yours.

5. Testing Nightmares

# Testing LangChain requires mocking framework internals
from unittest.mock import MagicMock, patch

def test_with_langchain():
    with patch("langchain_anthropic.chat_models.AnthropicLLM._call") as mock:
        mock.return_value = {"output": "4"}

        llm = ChatAnthropic()
        result = llm.invoke("What is 2+2?")

        assert result == "4"
        # This test mocks LangChain internals, not our logic

With a direct client, you test your code:

# Testing direct calls is straightforward
def test_direct_client():
    mock_client = Mock()
    mock_client.messages.create.return_value = MagicMock(
        content=[MagicMock(text="4")]
    )

    client = Anthropic(client=mock_client)
    result = client.complete("What is 2+2?")

    assert result == "4"
    # This tests our actual integration code

What I Built Instead: 15KB LLM Client

I needed 5 capabilities:

1. HTTP client for API calls
2. Streaming support
3. Circuit breaker
4. Token counting
5. Fallback chains

Total code: ~500 lines (15KB).

1. Minimal HTTP Client

import httpx
from typing import Optional, Dict, Any, AsyncGenerator
import json

class LLMClient:
    """Minimal LLM client with streaming and retry support."""

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.anthropic.com",
        timeout: float = 30.0
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=timeout)

    async def complete(
        self,
        prompt: str,
        system: Optional[str] = None,
        max_tokens: int = 1024,
        temperature: float = 1.0,
        model: str = "claude-3-5-sonnet-20241022"
    ) -> Dict[str, Any]:
        """Generate completion."""
        headers = self._build_headers()
        payload = self._build_payload(prompt, system, max_tokens, temperature, model)

        response = await self.client.post(
            f"{self.base_url}/v1/messages",
            headers=headers,
            json=payload
        )
        response.raise_for_status()

        return self._parse_response(response.json())

    def _build_headers(self) -> Dict[str, str]:
        return {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json"
        }

    def _build_payload(
        self,
        prompt: str,
        system: Optional[str],
        max_tokens: int,
        temperature: float,
        model: str
    ) -> Dict[str, Any]:
        payload = {
            "model": model,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "messages": [{"role": "user", "content": prompt}]
        }
        if system:
            payload["system"] = system
        return payload

    def _parse_response(self, response: Dict) -> Dict[str, Any]:
        return {
            "content": response["content"][0]["text"],
            "input_tokens": response["usage"]["input_tokens"],
            "output_tokens": response["usage"]["output_tokens"]
        }

2. Streaming Support

async def stream_complete(
    self,
    prompt: str,
    system: Optional[str] = None,
    max_tokens: int = 1024,
    model: str = "claude-3-5-sonnet-20241022"
) -> AsyncGenerator[str, None]:
    """Stream completion tokens."""
    headers = self._build_headers()
    payload = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": prompt}],
        "stream": True
    }
    if system:
        payload["system"] = system

    async with self.client.stream(
        "POST",
        f"{self.base_url}/v1/messages",
        headers=headers,
        json=payload
    ) as response:
        response.raise_for_status()

        async for line in response.aiter_lines():
            if line.startswith("data: "):
                data = json.loads(line[6:])

                if data["type"] == "content_block_delta":
                    yield data["delta"]["text"]
                elif data["type"] == "message_stop":
                    break

3. Circuit Breaker

from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    """Circuit breaker for external API calls."""

    def __init__(
        self,
        failure_threshold: int = 5,
        timeout_seconds: int = 60,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timedelta(seconds=timeout_seconds)
        self.success_threshold = success_threshold

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure = None

    def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection."""
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure > self.timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)

            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0

            return result

        except Exception as e:
            self.failure_count += 1
            self.last_failure = datetime.now()

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

            raise

4. Token Counting

import tiktoken

class TokenCounter:
    """Count tokens and estimate costs."""

    def __init__(self, model: str = "claude-3-5-sonnet-20241022"):
        self.model = model
        # Claude uses a similar tokenizer to GPT-4
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def count(self, text: str) -> int:
        """Count tokens in text."""
        return len(self.encoder.encode(text))

    def estimate_cost(
        self,
        input_tokens: int,
        output_tokens: int,
        model: str = None
    ) -> float:
        """Estimate API cost in USD."""
        model = model or self.model

        # Claude 3.5 Sonnet pricing
        pricing = {
            "claude-3-5-sonnet-20241022": (3.00, 15.00),  # per 1M input/output
            "claude-3-haiku-20241022": (0.25, 1.25),
            "claude-3-opus-20241022": (15.00, 75.00)
        }

        input_rate, output_rate = pricing.get(model, (3.00, 15.00))

        return (
            (input_tokens / 1_000_000) * input_rate +
            (output_tokens / 1_000_000) * output_rate
        )

5. Fallback Chains

class FallbackChain:
    """Try multiple models until one succeeds."""

    def __init__(self, clients: list[LLMClient]):
        self.clients = clients

    async def complete(self, prompt: str, **kwargs) -> Dict[str, Any]:
        """Try each client until one succeeds."""
        errors = []

        for client in self.clients:
            try:
                return await client.complete(prompt, **kwargs)
            except Exception as e:
                errors.append(f"{client.model}: {str(e)}")
                continue

        raise Exception(f"All clients failed: {errors}")

Results Comparison

Metric	LangChain	Direct (15KB)
Code size	15MB+	15KB
Dependencies	47 packages	2 (httpx, tiktoken)
Avg latency	420ms	165ms
Memory/request	12MB	3MB
Test coverage	61%	94%
Time to add model	4 hours	30 minutes

When LangChain Makes Sense

I'm not saying "never use LangChain." Valid use cases:

• Prototyping: Explore different approaches quickly
• Internal tools: Where reliability matters less
• Team familiarity: If your team is already trained on it
• Complex agents: If you actually need the agent abstractions

When to Go Direct

Consider the direct approach when:

• Performance matters: <500ms responses required
• Production reliability: 99.9%+ uptime needed
• Long-term products: Will evolve over 12+ months
• Full control: You want to control API calls
• Testing: You want to test your logic, not framework internals

The Code

GitHub: ChunkyTortoise/llm-integration-starter

Features:

• Streaming support
• Circuit breaker
• Fallback chains
• Token counting
• 149 tests, 94% coverage
• MIT licensed

Lessons Learned

1. Abstractions have costs. Every layer adds latency, memory, and debugging complexity.

2. APIs are simple. The Anthropic API is well-designed. Calling it directly is easier than learning a framework.

3. Dependencies are liabilities. Every dependency is code you don't control. Minimize them.

4. Test what you control. Testing your code is easy. Testing framework internals is a nightmare.

5. Profile before optimizing. I assumed the API was slow. It was LangChain.

Conclusion

LangChain adds 255ms overhead per request. I replaced it with 15KB of httpx. Now 3x faster, 94% test coverage, and zero dependency issues.

For production systems where latency and reliability matter, consider the direct approach.

---

Building AI infrastructure that actually works. Follow for more posts on production LLM engineering.