title: Building LLM Applications with LangChain — Production Guide description: LangChain production deployment guide: chain design, tool integration, memory management, evaluation, and debugging. Original tutorial with 5 production patterns. tags: [ai, langchain, llm, ops] lang: en date: 2026-10-23 original: true faq: - q: "How to debug LangChain chains in production?" a: "Enable verbose logging (verbose=True), use LangSmith for tracing, add try/except with detailed error context, and log input/output at each chain step." - q: "What are common LangChain memory pitfalls?" a: "Memory leaks from unbounded conversation history, stale context from old sessions, and token limit exceeded errors. Use WindowBufferMemory with max_tokens limit and implement session cleanup." - q: "How to handle LLM rate limits in LangChain apps?" a: "Implement exponential backoff with tenacity library, use request queuing, cache repeated queries, and set appropriate timeout values. Example: @retry(wait=wait_exponential(multiplier=1, min=4, max=60))" - q: "When to use LangChain vs building custom solution?" a: "Use LangChain for rapid prototyping and standard RAG patterns. Build custom for: complex orchestration, strict latency requirements, specialized tools, or when LangChain overhead is unacceptable." - q: "How to evaluate LangChain application quality?" a: "Use LangChain's evaluation framework: load_evaluator('qa') for answer relevance, load_evaluator('hallucination') for faithfulness. Build test sets with gold standard answers and track metrics over time."
Building LLM Applications with LangChain — Production Guide
Use Case: Build production-grade LLM applications using LangChain — chatbots, RAG systems, automated workflows, and AI agents.
1. LangChain Architecture Overview
LangChain provides a structured way to compose LLM applications from reusable components:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Prompts │ ──▶ │ LLM │ ──▶ │ Parsers │ ──▶ │ Output │
│ Templates │ │ (Ollama/ │ │ (JSON/ │ │ Validation │
│ │ │ OpenAI) │ │ Pydantic) │ │ │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
▲ │
│ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Tools │ ◀── │ Agent │ ◀── │ Memory │ ◀── │ Callbacks │
│ (APIs) │ │ (Orchestr.)│ │ (Context) │ │ (Logging) │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
2. Chain Design Patterns
2.1 Sequential Chain
from langchain.chains import SequentialChain
# Define individual chains
translation_chain = LLMChain(llm=llm, prompt=translate_prompt, output_key="translated")
summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")
# Combine into sequential chain
overall_chain = SequentialChain(
chains=[translation_chain, summary_chain],
input_variables=["text"],
output_variables=["translated", "summary"],
verbose=True
)
result = overall_chain({"text": "Original text to translate and summarize"})
2.2 RAG Chain
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.embeddings import OllamaEmbeddings
from langchain.llms import Ollama
# Setup
embeddings = OllamaEmbeddings(model="bge-m3", base_url="http://localhost:11434")
db = Chroma(persist_directory="/data/chroma/knowledge-base", embedding_function=embeddings)
llm = Ollama(model="qwen2.5:7b", base_url="http://localhost:11434")
# Create RAG chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=db.as_retriever(
search_kwargs={"k": 5, "score_threshold": 0.6}
),
return_source_documents=True,
chain_type="stuff" # Options: stuff, map_reduce, refine, map_rerank
)
response = qa_chain({"query": "How to troubleshoot Nginx 502 errors?"})
print(response["result"])
print(response["source_documents"])
2.3 Agent Chain
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
def check_service_status(service: str) -> str:
"""Check if a Linux service is running"""
import subprocess
result = subprocess.run(
["systemctl", "is-active", service],
capture_output=True, text=True
)
return result.stdout.strip()
tools = [
Tool(
name="check_service",
func=check_service_status,
description="Check if a Linux service is running"
)
]
agent = create_openai_functions_agent(llm=llm, tools=tools, prompt=agent_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "Is the nginx service running?"})
3. Tool Integration
3.1 API Tool Pattern
from langchain.tools import Tool
import aiohttp
async def query_prometheus(query: str) -> str:
"""Query Prometheus for metrics"""
async with aiohttp.ClientSession() as session:
async with session.get(
f"http://prometheus:9090/api/v1/query?query={query}"
) as resp:
return await resp.text()
prometheus_tool = Tool(
name="query_prometheus",
func=query_prometheus,
description="Query Prometheus metrics with PromQL"
)
3.2 Shell Command Tool (Sandboxed)
import asyncio
from langchain.tools import Tool
async def safe_shell(command: str, timeout: int = 30) -> str:
"""Execute shell command in sandbox"""
# Allowlist of safe commands
ALLOWED_PREFIXES = ["df ", "du ", "ps ", "top ", "grep ", "cat ", "systemctl status "]
if not any(command.startswith(prefix) for prefix in ALLOWED_PREFIXES):
return "Error: Command not allowed"
try:
process = await asyncio.create_subprocess_exec(
*command.split(),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
return stdout.decode() if stdout else stderr.decode()
except asyncio.TimeoutError:
return "Error: Command timed out"
shell_tool = Tool(
name="safe_shell",
func=safe_shell,
description="Execute safe shell commands for diagnostics"
)
4. Memory Management
4.1 Conversation History
from langchain.memory import ConversationBufferMemory, WindowBufferMemory
# Unbounded memory (risk of token overflow)
buffer_memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# Bounded memory (recommended for production)
window_memory = WindowBufferMemory(
memory_key="chat_history",
return_messages=True,
k=10 # Keep last 10 messages
)
4.2 Persistent Memory with Vector Store
from langchain.vectorstores import Chroma
from langchain.storage import InMemoryStore
from langchain.document_loaders import ChatOpenAI
# Persistent conversation storage
store = InMemoryStore()
vectorstore = Chroma(persist_directory="/data/langchain/memory", embedding_function=embeddings)
# Load previous conversations
def load_conversation(session_id: str) -> list:
docs = vectorstore.similarity_search(session_id, k=5)
return [doc.page_content for doc in docs]
5. Evaluation and Testing
5.1 Answer Relevance Evaluation
from langchain.evaluation import load_evaluator
# Load evaluator
relevance_evaluator = load_evaluator("qa")
def evaluate_answer(question: str, answer: str) -> dict:
"""Evaluate answer relevance"""
result = relevance_evaluator.evaluate_strings(
input=question,
prediction=answer
)
return {
"score": result["score"],
"reasoning": result.get("reasoning", "")
}
5.2 Hallucination Detection
from langchain.evaluation import load_evaluator
hallucination_evaluator = load_evaluator("labeled_hallucination")
def detect_hallucination(question: str, context: str, answer: str) -> dict:
"""Detect hallucinations in LLM output"""
result = hallucination_evaluator.evaluate_strings(
prediction=answer,
input=question,
reference=context
)
return {
"is_hallucinated": result["score"] < 0.5,
"confidence": result["score"]
}
5.3 Batch Evaluation
TEST_CASES = [
{
"input": "What is the CPU usage of node-1?",
"expected_keywords": ["cpu", "usage", "percentage"]
},
{
"input": "List all running Docker containers",
"expected_keywords": ["container", "running", "docker"]
}
]
def batch_evaluate(chain, test_cases):
results = []
for case in test_cases:
output = chain.invoke({"input": case["input"]})
keywords_found = all(kw in output.lower() for kw in case["expected_keywords"])
results.append({
"input": case["input"],
"output": output,
"pass": keywords_found
})
return results
6. Production Debugging
6.1 Verbose Logging
# Enable verbose mode for debugging
llm = Ollama(model="qwen2.5:7b", base_url="http://localhost:11434", verbose=True)
# Or set environment variable
import os
os.environ["LANGCHAIN_VERBOSE"] = "true"
6.2 LangSmith Tracing
import langchain
langchain.debug = True
# Or use LangSmith for production tracing
from langsmith import Client
client = Client()
run_id = client.trace(
name="my_chain",
inputs={"query": "test"},
outputs={"result": "answer"}
)
6.3 Error Handling
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
@retry(
wait=wait_exponential(multiplier=1, min=2, max=30),
stop=stop_after_attempt(3),
retry=retry_if_exception_type((TimeoutError, Exception))
)
def robust_chain_invoke(chain, input_data):
"""Chain invocation with retry logic"""
try:
return chain.invoke(input_data)
except Exception as e:
logger.error(f"Chain failed: {e}")
raise
7. Performance Optimization
7.1 Streaming Responses
from langchain.callbacks import StreamingCallbackHandler
class TokenStreamingCallback(StreamingCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs):
yield token # Send tokens as they're generated
# Use streaming
for token in llm.stream("Generate a summary"):
print(token, end="", flush=True)
7.2 Caching
from langchain.cache import SQLiteCache
import langchain
# Enable caching
langchain.vectorstore.cache = SQLiteCache(database_path="/tmp/langchain.db")
# Or use in-memory cache for development
langchain.vectorstore.cache = InMemoryCache()