Langchain Reference Architecture

src/my_service/
├── app/                         # Layer 1: HTTP boundary (FastAPI)
│   ├── __init__.py
│   ├── main.py                  # FastAPI instance, DI wiring, lifespan
│   ├── routes/
│   │   ├── support.py           # POST /support → services.support.run(...)
│   │   └── health.py
│   └── deps.py                  # FastAPI Depends() providers
├── services/                    # Layer 2: chain and graph definitions
│   ├── __init__.py
│   ├── registry.py              # name → builder lookup
│   ├── support/
│   │   ├── chain.py             # SupportChain(llm, retriever, memory)
│   │   └── graph.py             # SupportGraph (LangGraph StateGraph)
│   └── triage/
│       └── chain.py
├── adapters/                    # Layer 3: vendor integrations
│   ├── __init__.py
│   ├── llm_factory.py           # chat_model(provider, **kwargs) → BaseChatModel
│   ├── retriever_factory.py     # retriever_for(tenant_id) → Retriever
│   ├── tool_factory.py          # tools_for(tenant_id) → list[BaseTool]
│   ├── checkpointer.py          # checkpointer_for(env) → BaseCheckpointSaver
│   └── history.py               # history_for(session_id, tenant_id) → BaseChatMessageHistory
├── config/                      # Layer 4: configuration
│   ├── __init__.py
│   └── settings.py              # Pydantic Settings
└── domain/                      # Layer 5: pure models, no I/O
    ├── __init__.py
    ├── state.py                 # TypedDict / Pydantic for LangGraph state
    └── models.py                # request/response schemas
tests/
├── unit/                        # fake adapters, assert service logic
├── integration/                 # real adapters against ephemeral infra
└── contract/                    # schema snapshots (e.g., tool specs)
pyproject.toml                   # includes [tool.importlinter] contracts

# src/my_service/adapters/llm_factory.py
from langchain_core.language_models import BaseChatModel
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI

_SAFE_DEFAULTS = {"timeout": 30, "max_retries": 2}

def chat_model(provider: str, **overrides) -> BaseChatModel:
    defaults = {**_SAFE_DEFAULTS, **overrides}  # caller wins
    if provider == "anthropic":
        return ChatAnthropic(model="claude-sonnet-4-6", **defaults)
    if provider == "openai":
        return ChatOpenAI(model="gpt-4o", **defaults)
    raise ValueError(f"Unknown provider: {provider!r}")

# src/my_service/services/registry.py
from typing import Callable, Protocol
from langchain_core.runnables import Runnable

class ChainBuilder(Protocol):
    def __call__(self, *, tenant_id: str) -> Runnable: ...

_BUILDERS: dict[str, ChainBuilder] = {}

def register(name: str):
    def decorator(fn: ChainBuilder) -> ChainBuilder:
        _BUILDERS[name] = fn
        return fn
    return decorator

def get(name: str, *, tenant_id: str) -> Runnable:
    try:
        return _BUILDERSname
    except KeyError:
        raise KeyError(f"No chain registered under {name!r}. Known: {list(_BUILDERS)}")

# src/my_service/services/support/chain.py
from my_service.services.registry import register
from my_service.adapters.llm_factory import chat_model
from my_service.adapters.retriever_factory import retriever_for

@register("support_agent")
def build_support_agent(*, tenant_id: str):
    llm = chat_model("anthropic")
    retriever = retriever_for(tenant_id=tenant_id)
    # ... compose chain ...
    return chain

# src/my_service/adapters/retriever_factory.py
from functools import lru_cache
from langchain_core.retrievers import BaseRetriever
from langchain_pinecone import PineconeVectorStore
from my_service.config.settings import get_settings

@lru_cache(maxsize=256)  # cache the *store*, not the retriever
def _store_for(tenant_id: str) -> PineconeVectorStore:
    s = get_settings()
    return PineconeVectorStore(
        index_name=s.pinecone_index,
        namespace=f"tenant:{tenant_id}",  # per-tenant namespace
        embedding=...,
    )

def retriever_for(*, tenant_id: str, k: int = 6) -> BaseRetriever:
    # Retriever construction <5ms because store is cached — do it per-request.
    return _store_for(tenant_id).as_retriever(search_kwargs={"k": k})

# src/my_service/config/settings.py
from functools import lru_cache
from typing import Literal
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="MYSVC_")

    env: Literal["dev", "staging", "prod"] = "dev"
    anthropic_api_key: SecretStr
    openai_api_key: SecretStr
    pinecone_api_key: SecretStr
    pinecone_index: str
    postgres_dsn: SecretStr | None = None  # required when env != "dev"

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()  # reads env/.env at first call, caches

# src/my_service/adapters/middleware.py
from langchain_core.runnables import Runnable

def wrap(model: Runnable) -> Runnable:
    # Order matters: redact -> cache -> retry -> model
    # Cross-reference L31 (langchain-middleware-patterns) for the full rationale.
    return (
        model
        .with_config(tags=["mysvc"])
        # | redaction_middleware()
        # | cache_middleware()
        # | retry_middleware()
    )

# src/my_service/adapters/checkpointer.py
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.checkpoint.memory import MemorySaver

def checkpointer_for(env: str) -> BaseCheckpointSaver:
    if env == "dev":
        return MemorySaver()
    # Staging/prod: Postgres-backed. Async variant for FastAPI.
    from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
    from my_service.config.settings import get_settings
    dsn = get_settings().postgres_dsn
    assert dsn is not None, "POSTGRES_DSN required outside dev"
    return AsyncPostgresSaver.from_conn_string(dsn.get_secret_value())

# tests/unit/test_support_chain.py
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from my_service.services.support.chain import build_support_agent

def test_support_agent_returns_expected_shape(monkeypatch):
    monkeypatch.setattr(
        "my_service.services.support.chain.chat_model",
        lambda provider, **kw: FakeListChatModel(responses=["fixed answer"]),
    )
    chain = build_support_agent(tenant_id="acme")
    assert chain.invoke({"input": "hi"}).content == "fixed answer"