Beyond Basic LangChain
You've built a simple agent. Now scale it.
Production LangChain systems require:
- Memory management
- Error handling
- Performance optimization
- Monitoring & observability
Memory Patterns
Conversation Memory
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
agent = initialize_agent(
tools, llm, memory=memory, agent_type="conversational"
)
Summary Memory (for long conversations)
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(
llm=OpenAI(),
buffer="Current conversation summarized"
)
Tool Chains & Sequences
Sequential Chain
from langchain.chains import SequentialChain
chain = SequentialChain(
chains=[chain1, chain2, chain3],
verbose=True
)
Conditional Routing
router_template = """Given the input, route to:
analysis, coding, or research
Input: {input}
Route:"""
router = llm_chain.run(router_template)
if "coding" in router:
result = coding_agent.run(input)
Error Handling & Retry Logic
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))def safe_agent_run(query):
return agent.run(query)
try:
result = safe_agent_run(query)
except Exception as e:
logger.error(f"Agent failed: {e}")
result = fallback_response()
Performance Optimization
Caching
from langchain.cache import RedisCache
import redis
redis_client = redis.Redis.from_url("redis://localhost")
langchain.llm_cache = RedisCache(redis_client=redis_client)
Batch Processing
results = [agent.run(q) for q in queries]
# Better: Use async
import asyncio
results = await asyncio.gather(*[async_agent(q) for q in queries])
Monitoring & Observability
import logging
from datetime import datetime
class AgentLogger:
def log_run(self, query, response, duration):
logging.info(f"Query: {query}")
logging.info(f"Response: {response}")
logging.info(f"Duration: {duration}s")
# Track metrics
self.track_metric("agent_latency", duration)
self.track_metric("token_usage", count_tokens(response))
Integration with Vector Stores
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vector_store = Pinecone.from_documents(docs, embeddings)
retriever = vector_store.as_retriever()
agent_with_retrieval = RetrievalQA.from_chain_type(
llm=llm,
retriever=retriever
)
Deployment Strategies
Local + Cloud Hybrid
- Local cache for frequently used data
- Cloud for complex reasoning
- Best of both worlds
Serverless Deployment
# AWS Lambda
def lambda_handler(event, context):
query = event['query']
result = agent.run(query)
return {'statusCode': 200, 'body': result}
Testing Your Agent
def test_agent_accuracy():
test_cases = [
("query1", "expected_output1"),
("query2", "expected_output2")
]
for query, expected in test_cases:
result = agent.run(query)
assert verify_correctness(result, expected)
Production Checklist
✅ Error handling for all tool calls
✅ Logging for debugging
✅ Monitoring & alerting
✅ Rate limiting
✅ Input validation
✅ Output sanitization
✅ Cost tracking
✅ Performance metrics
✅ Rollback procedures
✅ Security hardening
Common Production Issues
Issue 1: Token limits exceeded
→ Solution: Summarize long conversations
Issue 2: Tool calls fail silently
→ Solution: Add explicit error messages
Issue 3: Costs spiral out of control
→ Solution: Implement token budgets
Issue 4: Model drift over time
→ Solution: Regular monitoring & retraining
The Enterprise Path
LangChain in enterprise = structured, monitored, optimized.
You now have the patterns to build production systems.
What LangChain patterns are you using?
