方寸 Portal

AI Engineer Portal

Your personal operating system for career transition.

Private mode
Back to Learning Paths
Lesson

Provider abstraction patterns

Compare LangChain's model interface against direct SDK calls, understand the callback system, and decide when abstraction pays off.

40 min
llm-app-foundationsphase-1portfolio

Why this matters

Every AI team eventually faces the same question: do we wrap the provider SDK, or use it directly? The answer changes the codebase's testability, portability, and operational complexity. LangChain's model interface is one answer. Rolling your own thin wrapper is another. Knowing the tradeoffs — and the concrete costs of each — is the kind of architectural judgment that distinguishes senior AI engineers.

Core concepts

LangChain's model interface vs rolling your own

LangChain's chat model interface normalizes the provider API to a single method signature:

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import HumanMessage, AIMessage

# Works identically for Anthropic, OpenAI, Google, local models
def generate_response(llm: BaseChatModel, question: str) -> str:
    messages = [HumanMessage(content=question)]
    response = llm.invoke(messages)
    return response.content

Your own thin wrapper looks like:

from typing import Protocol

class LLMClient(Protocol):
    def generate(self, prompt: str, system: str = "") -> str: ...

class AnthropicClient:
    def __init__(self, model: str = "claude-3-5-sonnet-20241022"):
        import anthropic
        self._client = anthropic.Anthropic()
        self._model = model

    def generate(self, prompt: str, system: str = "") -> str:
        response = self._client.messages.create(
            model=self._model,
            max_tokens=1024,
            system=system,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text

The Protocol approach is simpler, has zero external dependencies, and is easier to mock in tests. LangChain's interface is richer but requires understanding its message types, invoke() vs stream() vs batch(), and how it handles tool calls.

When to use LangChain vs direct SDK calls

SituationRecommendation
Single provider, simple promptsDirect SDK calls
Multiple providers, same prompt logicLangChain model abstraction
Complex chains with branchingLCEL
Need streaming to the UIDirect SDK (simpler control)
Structured output with PydanticLangChain .with_structured_output()
Rapid prototypingEither; LangChain has more batteries
Production service, team unfamiliar with LCDirect SDK + thin wrapper

Callback system for logging and tracing

LangChain's callback system is one of its most underrated features. Callbacks fire at every step of a chain execution: on start, on token, on end, on error.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
import time
import logging

logger = logging.getLogger(__name__)

class LatencyAndCostCallback(BaseCallbackHandler):
    def __init__(self):
        self._start_times: dict[str, float] = {}

    def on_llm_start(self, serialized: dict, prompts: list[str], **kwargs) -> None:
        run_id = str(kwargs.get("run_id", "unknown"))
        self._start_times[run_id] = time.monotonic()
        logger.info("llm_start", extra={"model": serialized.get("name"), "run_id": run_id})

    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        run_id = str(kwargs.get("run_id", "unknown"))
        start = self._start_times.pop(run_id, time.monotonic())
        latency_ms = int((time.monotonic() - start) * 1000)
        usage = response.llm_output.get("usage", {}) if response.llm_output else {}
        logger.info(
            "llm_end",
            extra={
                "run_id": run_id,
                "latency_ms": latency_ms,
                "input_tokens": usage.get("input_tokens", 0),
                "output_tokens": usage.get("output_tokens", 0),
            },
        )

    def on_llm_error(self, error: Exception, **kwargs) -> None:
        logger.error("llm_error", extra={"error": str(error)})

# Attach to any chain
from langchain_anthropic import ChatAnthropic
callback = LatencyAndCostCallback()
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", callbacks=[callback])

This is the LangChain equivalent of middleware. You get observability without polluting your business logic.

Same task: LangChain vs bare Anthropic SDK

Here is a support ticket classifier implemented both ways, so you can see the real tradeoffs.

With LangChain:

from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

class TicketResult(BaseModel):
    category: str = Field(description="billing, technical, account, or feature_request")
    priority: str = Field(description="low, medium, high, or urgent")

def classify_with_langchain(ticket: str) -> TicketResult:
    llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
    structured = llm.with_structured_output(TicketResult)
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Classify this support ticket."),
        ("human", "{ticket}"),
    ])
    chain = prompt | structured
    return chain.invoke({"ticket": ticket})

With bare Anthropic SDK:

import anthropic
import json
from pydantic import BaseModel, Field

class TicketResult(BaseModel):
    category: str = Field(description="billing, technical, account, or feature_request")
    priority: str = Field(description="low, medium, high, or urgent")

def classify_with_sdk(ticket: str) -> TicketResult:
    client = anthropic.Anthropic()
    tool_schema = {
        "name": "classify_ticket",
        "description": "Classify a support ticket",
        "input_schema": TicketResult.model_json_schema(),
    }
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=256,
        system="Classify this support ticket.",
        messages=[{"role": "user", "content": ticket}],
        tools=[tool_schema],
        tool_choice={"type": "tool", "name": "classify_ticket"},
    )
    tool_use = next(b for b in response.content if b.type == "tool_use")
    return TicketResult.model_validate(tool_use.input)

Tradeoffs:

LangChainBare SDK
Lines of code~10~18
Dependencieslangchain + langchain-anthropicanthropic
Stack trace clarityPoor (many wrappers)Excellent
Provider swapOne lineRewrite tool schema
Callback/tracingBuilt-inManual
DebuggabilityHarderEasier

For a single-provider internal tool: bare SDK. For a multi-provider product feature: LangChain.

Common mistakes

Mixing LangChain message types with raw dicts. LangChain uses HumanMessage, AIMessage, SystemMessage objects. The raw Anthropic SDK uses dicts with role and content keys. They are not interchangeable. Pick one layer and stay there.

Using LangChain callbacks for business logic. Callbacks are for observability (logging, tracing, metrics). Putting business logic in an on_llm_end callback creates hidden side effects that are extremely hard to test.

Not setting verbose=False in production. LangChain defaults to printing chain execution to stdout. In production services, this floods logs. Always initialize with verbose=False or configure logging properly.

Assuming LangChain handles retries. It does not retry on provider rate limits by default. Add retry logic at the HTTP client level or use a library like tenacity regardless of which abstraction you use.

Try it yourself

Implement the same task — extracting a MeetingNotes Pydantic model with attendees: list[str], action_items: list[str], and decisions: list[str] from a raw meeting transcript — using both LangChain and the bare Anthropic SDK. Time both implementations on 10 transcripts. Measure lines of code, import time, and total latency. Write down which you would choose for a production feature and why.

Lesson Deep Dive

Ask a follow-up question about this lesson and get an AI-powered explanation.

Loading previous questions...