tomevault-io LangChain Reference Architecture
Overview
Production-ready architectural patterns for building scalable, maintainable LangChain applications.
Prerequisites
- Understanding of LangChain fundamentals
- Experience with software architecture
- Knowledge of cloud infrastructure
Architecture Patterns
Pattern 1: Layered Architecture
src/
├── api/ # API layer (FastAPI/Flask)
│ ├── __init__.py
│ ├── routes/
│ │ ├── chat.py
│ │ └── agents.py
│ └── middleware/
│ ├── auth.py
│ └── rate_limit.py
├── core/ # Business logic layer
│ ├── __init__.py
│ ├── chains/
│ │ ├── __init__.py
│ │ ├── chat_chain.py
│ │ └── rag_chain.py
│ ├── agents/
│ │ ├── __init__.py
│ │ └── research_agent.py
│ └── tools/
│ ├── __init__.py
│ └── search.py
├── infrastructure/ # Infrastructure layer
│ ├── __init__.py
│ ├── llm/
│ │ ├── __init__.py
│ │ └── provider.py
│ ├── vectorstore/
│ │ └── pinecone.py
│ └── cache/
│ └── redis.py
├── config/ # Configuration
│ ├── __init__.py
│ └── settings.py
└── main.py
Pattern 2: Provider Abstraction
# infrastructure/llm/provider.py
from abc import ABC, abstractmethod
from langchain_core.language_models import BaseChatModel
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
class LLMProvider(ABC):
"""Abstract LLM provider."""
@abstractmethod
def get_chat_model(self, **kwargs) -> BaseChatModel:
pass
class OpenAIProvider(LLMProvider):
def get_chat_model(self, model: str = "gpt-4o-mini", **kwargs) -> BaseChatModel:
return ChatOpenAI(model=model, **kwargs)
class AnthropicProvider(LLMProvider):
def get_chat_model(self, model: str = "claude-3-5-sonnet-20241022", **kwargs) -> BaseChatModel:
return ChatAnthropic(model=model, **kwargs)
class LLMFactory:
"""Factory for creating LLM instances."""
_providers = {
"openai": OpenAIProvider(),
"anthropic": AnthropicProvider(),
}
@classmethod
def create(cls, provider: str = "openai", **kwargs) -> BaseChatModel:
if provider not in cls._providers:
raise ValueError(f"Unknown provider: {provider}")
return cls._providers[provider].get_chat_model(**kwargs)
# Usage
llm = LLMFactory.create("openai", model="gpt-4o-mini")
Pattern 3: Chain Registry
# core/chains/__init__.py
from typing import Dict, Type
from langchain_core.runnables import Runnable
class ChainRegistry:
"""Registry for managing chains."""
_chains: Dict[str, Runnable] = {}
@classmethod
def register(cls, name: str, chain: Runnable) -> None:
cls._chains[name] = chain
@classmethod
def get(cls, name: str) -> Runnable:
if name not in cls._chains:
raise ValueError(f"Chain '{name}' not found")
return cls._chains[name]
@classmethod
def list_chains(cls) -> list:
return list(cls._chains.keys())
# Register chains at startup
from core.chains.chat_chain import create_chat_chain
from core.chains.rag_chain import create_rag_chain
ChainRegistry.register("chat", create_chat_chain())
ChainRegistry.register("rag", create_rag_chain())
# Usage in API
@app.post("/invoke/{chain_name}")
async def invoke_chain(chain_name: str, request: InvokeRequest):
chain = ChainRegistry.get(chain_name)
return await chain.ainvoke(request.input)
Pattern 4: RAG Architecture
# core/chains/rag_chain.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_pinecone import PineconeVectorStore
def create_rag_chain(
llm: BaseChatModel = None,
vectorstore: VectorStore = None
) -> Runnable:
"""Create a RAG chain with retrieval."""
llm = llm or ChatOpenAI(model="gpt-4o-mini")
vectorstore = vectorstore or PineconeVectorStore.from_existing_index(
index_name="knowledge-base",
embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
prompt = ChatPromptTemplate.from_template("""
Answer the question based on the following context:
Context:
{context}
Question: {question}
Answer:
""")
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
)
return chain
Pattern 5: Multi-Agent System
# core/agents/orchestrator.py
from langchain_core.runnables import RunnableLambda
from typing import Dict, Any
class AgentOrchestrator:
"""Orchestrate multiple specialized agents."""
def __init__(self):
self.agents = {}
self.router = None
def register_agent(self, name: str, agent: Runnable) -> None:
self.agents[name] = agent
def set_router(self, router: Runnable) -> None:
"""Set routing logic for agent selection."""
self.router = router
async def route_and_execute(self, input_data: Dict[str, Any]) -> Any:
"""Route input to appropriate agent and execute."""
# Determine which agent to use
agent_name = await self.router.ainvoke(input_data)
if agent_name not in self.agents:
raise ValueError(f"Agent '{agent_name}' not found")
# Execute with selected agent
agent = self.agents[agent_name]
return await agent.ainvoke(input_data)
# Setup
orchestrator = AgentOrchestrator()
orchestrator.register_agent("research", research_agent)
orchestrator.register_agent("coding", coding_agent)
orchestrator.register_agent("general", general_agent)
# Router uses LLM to classify request
router_prompt = ChatPromptTemplate.from_template("""
Classify this request into one of: research, coding, general
Request: {input}
Classification:
""")
orchestrator.set_router(router_prompt | llm | StrOutputParser())
Pattern 6: Configuration-Driven Design
# config/settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
class LLMSettings(BaseSettings):
provider: str = "openai"
model: str = "gpt-4o-mini"
temperature: float = 0.7
max_tokens: int = 4096
max_retries: int = 3
class VectorStoreSettings(BaseSettings):
provider: str = "pinecone"
index_name: str = "default"
embedding_model: str = "text-embedding-3-small"
class Settings(BaseSettings):
llm: LLMSettings = Field(default_factory=LLMSettings)
vectorstore: VectorStoreSettings = Field(default_factory=VectorStoreSettings)
redis_url: str = "redis://localhost:6379"
log_level: str = "INFO"
class Config:
env_file = ".env"
env_nested_delimiter = "__"
settings = Settings()
# Usage
llm = LLMFactory.create(
settings.llm.provider,
model=settings.llm.model,
temperature=settings.llm.temperature
)
Architecture Diagram
┌─────────────────┐
│ API Gateway │
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌─────────▼─────┐ ┌──────▼──────┐ ┌─────▼─────────┐
│ Chat Chain │ │ RAG Chain │ │ Agent System │
└───────┬───────┘ └──────┬──────┘ └───────┬───────┘
│ │ │
└────────────────┼────────────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌─────────▼─────┐ ┌──────▼──────┐ ┌─────▼─────────┐
│ LLM Provider │ │ VectorStore │ │ Cache │
└───────────────┘ └─────────────┘ └───────────────┘
Output
- Layered architecture with clear separation
- Provider abstraction for LLM flexibility
- Chain registry for runtime management
- Multi-agent orchestration pattern
Resources
Next Steps
Use langchain-multi-env-setup for environment management.
Source: HelixDevelopment/HelixAgent — distributed by TomeVault.