---
allowed-tools: run_command, view_file, write_to_file, list_dir
command: ai:langchain-optimize
description: "Optimize LangChain applications with Context7-verified LCEL patterns, async operations, caching, and RAG optimization strategies."

---

# langchain:optimize

Optimize LangChain applications with Context7-verified LCEL patterns, async operations, caching, and RAG optimization strategies.

## Description

Comprehensive LangChain optimization following official best practices:
- LangChain Expression Language (LCEL) patterns
- Async/streaming operations
- Chain composition and optimization
- Memory and caching strategies
- RAG pipeline optimization
- Vector store performance tuning
- Agent and tool optimization

## Required Documentation Access

**MANDATORY:** Before optimization, query Context7 for LangChain best practices:

**Documentation Queries:**
- `mcp://context7/langchain-ai/langchain` - LangChain core documentation
- `mcp://context7/websites/python_langchain` - Python LangChain patterns
- `mcp://context7/langchain/lcel-patterns` - LCEL composition patterns
- `mcp://context7/langchain/async-streaming` - Async and streaming optimization
- `mcp://context7/langchain/caching-strategies` - Cache optimization
- `mcp://context7/langchain/rag-optimization` - RAG pipeline tuning

**Why This is Required:**
- Ensures optimization follows official LangChain documentation
- Applies proven LCEL composition patterns
- Validates async and streaming strategies
- Prevents performance bottlenecks
- Optimizes memory and cache usage
- Implements best practices for production RAG

## Usage

```bash
/langchain:optimize [options]
```

## Options

- `--scope <lcel|async|caching|rag|agents|all>` - Optimization scope (default: all)
- `--analyze-only` - Analyze without applying changes
- `--output <file>` - Write optimization report

## Examples

### Full LangChain Optimization
```bash
/langchain:optimize
```

### LCEL Patterns Only
```bash
/langchain:optimize --scope lcel
```

### RAG Optimization
```bash
/langchain:optimize --scope rag
```

### Analyze Current System
```bash
/langchain:optimize --analyze-only --output langchain-report.md
```

## Optimization Categories

### 1. LCEL Pattern Optimization (Context7-Verified)

**Pattern from Context7 (/websites/python_langchain):**

#### Basic LCEL Chain
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# LCEL chain composition (declarative, optimized)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

model = ChatOpenAI(model="gpt-4o-mini")

output_parser = StrOutputParser()

# Chain with | operator (LCEL)
chain = prompt | model | output_parser

# Invoke chain
result = chain.invoke({"input": "What is LangChain?"})
print(result)
```

**Benefits of LCEL:**
- Automatic async/streaming support
- Built-in retry and fallback logic
- Optimized parallel execution
- Type safety and validation
- Easier debugging and monitoring

#### Complex LCEL Chain with Branching
```python
from langchain_core.runnables import RunnableBranch, RunnablePassthrough

# Conditional routing based on input
branch = RunnableBranch(
    (
        lambda x: "code" in x["topic"],
        ChatPromptTemplate.from_template("Explain this code: {input}") | model
    ),
    (
        lambda x: "math" in x["topic"],
        ChatPromptTemplate.from_template("Solve this problem: {input}") | model
    ),
    ChatPromptTemplate.from_template("Answer: {input}") | model  # Default
)

# Chain with routing
chain = (
    RunnablePassthrough.assign(topic=lambda x: x["input"].lower())
    | branch
    | StrOutputParser()
)

result = chain.invoke({"input": "Explain this Python code: def f(x): return x*2"})
# Routes to code explanation prompt
```

**Performance Impact:**
- LCEL chains: Automatic optimization and parallelization
- Manual chains: Require explicit async handling
- LCEL: 3x faster with parallel operations

#### LCEL with Fallbacks and Retries
```python
from langchain_core.runnables import RunnableWithFallbacks

# Primary and fallback models
primary = ChatOpenAI(model="gpt-4o", temperature=0)
fallback = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Chain with automatic fallback
chain_with_fallback = (
    prompt
    | primary.with_fallbacks([fallback])
    | StrOutputParser()
)

# Automatically falls back to gpt-4o-mini if gpt-4o fails
result = chain_with_fallback.invoke({"input": "Explain AI"})
```

**Benefits:**
- Automatic failover to cheaper model
- Built-in retry logic
- 99.9% uptime vs 95% without fallback
- Cost optimization (use expensive model only when needed)

### 2. Async and Streaming Optimization (Context7-Verified)

**Pattern from Context7 (/websites/python_langchain):**

#### Async Chain Execution
```python
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Async chain
async def async_chain_example():
    prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
    model = ChatOpenAI(model="gpt-4o-mini")

    chain = prompt | model

    # Async invoke
    result = await chain.ainvoke({"topic": "quantum computing"})
    print(result.content)

asyncio.run(async_chain_example())
```

#### Async Batch Processing
```python
# Process multiple inputs concurrently
async def batch_process(topics: list[str]):
    prompt = ChatPromptTemplate.from_template("Explain {topic} in one sentence")
    model = ChatOpenAI(model="gpt-4o-mini")

    chain = prompt | model | StrOutputParser()

    # Batch invoke (parallel execution)
    results = await chain.abatch([{"topic": t} for t in topics])
    return results

topics = ["AI", "ML", "DL", "NLP", "CV"]
results = asyncio.run(batch_process(topics))

# Performance:
# Sequential: 5 requests × 2s = 10s
# Async batch: max(5 × 2s) = 2s (5x faster)
```

**Performance Impact:**
- Sequential: O(n) time complexity
- Async batch: O(1) time complexity (up to rate limits)
- 5-10x speedup for multiple requests

#### Streaming Responses
```python
# Stream tokens as they arrive
def stream_example():
    prompt = ChatPromptTemplate.from_template("Write an essay about {topic}")
    model = ChatOpenAI(model="gpt-4o", streaming=True)

    chain = prompt | model

    # Stream response
    for chunk in chain.stream({"topic": "artificial intelligence"}):
        print(chunk.content, end="", flush=True)

# Async streaming
async def async_stream_example():
    prompt = ChatPromptTemplate.from_template("Explain {topic}")
    model = ChatOpenAI(model="gpt-4o", streaming=True)

    chain = prompt | model

    # Async stream
    async for chunk in chain.astream({"topic": "machine learning"}):
        print(chunk.content, end="", flush=True)

asyncio.run(async_stream_example())
```

**Benefits:**
- Time to first token: 500ms vs 5s for full response
- Better UX with progressive rendering
- Lower perceived latency (10x improvement)

### 3. Caching Strategies (Context7-Verified)

**Pattern from Context7 (/websites/python_langchain):**

#### In-Memory Cache
```python
from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache

# Enable in-memory caching
set_llm_cache(InMemoryCache())

model = ChatOpenAI(model="gpt-4o-mini")

# First call: API request
result1 = model.invoke("What is Python?")  # 2s

# Second call: Cached (instant)
result2 = model.invoke("What is Python?")  # <1ms
```

**Performance Impact:**
- First call: 2s (API request)
- Cached calls: <1ms (2000x faster)
- 100% cost savings for cached queries

#### Redis Cache for Production
```python
from langchain.cache import RedisCache
from redis import Redis

# Redis-backed cache
redis_client = Redis(host='localhost', port=6379)
set_llm_cache(RedisCache(redis_client))

model = ChatOpenAI(model="gpt-4o-mini")

# Cache shared across all server instances
result = model.invoke("Explain AI")  # Cached if queried before
```

**Benefits:**
- Persistent cache across restarts
- Shared cache across multiple servers
- TTL support for automatic expiration
- Production-ready scalability

#### Semantic Cache
```python
from langchain.cache import RedisSemanticCache
from langchain_openai import OpenAIEmbeddings

# Cache based on semantic similarity
embeddings = OpenAIEmbeddings()

set_llm_cache(
    RedisSemanticCache(
        redis_url="redis://localhost:6379",
        embedding=embeddings,
        score_threshold=0.2  # Cache hit if similarity > 0.8
    )
)

model = ChatOpenAI(model="gpt-4o-mini")

# These queries are semantically similar, second uses cache
result1 = model.invoke("What is artificial intelligence?")
result2 = model.invoke("Can you explain AI?")  # Cache hit (similar meaning)
```

**Benefits:**
- Matches semantically similar queries
- 40-60% cache hit rate vs 20% with exact matching
- Reduces costs even with query variations

### 4. RAG Optimization (Context7-Verified)

**Pattern from Context7 (/websites/python_langchain):**

#### Optimized RAG Chain
```python
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Setup vector store with cached embeddings
from langchain_community.embeddings import CacheBackedEmbeddings
from langchain_community.storage import RedisStore

store = RedisStore(redis_url="redis://localhost:6379")
underlying_embeddings = OpenAIEmbeddings()

cached_embedder = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    store,
    namespace="openai_embeddings"
)

# Create vector store
vector_store = FAISS.from_documents(documents, cached_embedder)

# MMR retriever for diversity
retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,
        "lambda_mult": 0.7
    }
)

# RAG prompt
template = """Answer the question based on the following context:

Context: {context}

Question: {question}

Answer:"""

prompt = ChatPromptTemplate.from_template(template)

# RAG chain with LCEL
rag_chain = (
    {
        "context": retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])),
        "question": RunnablePassthrough()
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# Query
result = rag_chain.invoke("What is machine learning?")
```

**Performance Optimizations:**
- Cached embeddings: 59x faster (17ms vs 1s)
- MMR retrieval: 40% better diversity
- FAISS vector store: 50x faster than linear search
- LCEL composition: Automatic parallelization

#### Multi-Query RAG
```python
from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)

# Generate multiple queries from single query
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vector_store.as_retriever(),
    llm=llm
)

# Enhanced RAG chain
rag_chain = (
    {
        "context": multi_query_retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])),
        "question": RunnablePassthrough()
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

result = rag_chain.invoke("What is deep learning?")
# Generates 3-5 queries, retrieves for each, merges results
```

**Benefits:**
- 50% better retrieval coverage
- Handles query ambiguity
- Multiple perspectives on the question
- Better answer quality

### 5. Agent Optimization (Context7-Verified)

**Pattern from Context7 (/websites/python_langchain):**

#### Optimized Agent with Tool Caching
```python
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# Define tools with caching
@tool
def get_weather(location: str) -> str:
    """Get current weather for a location."""
    # Implementation here
    return f"Weather in {location}: Sunny, 72°F"

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Implementation here
    return f"Search results for: {query}"

tools = [get_weather, search_web]

# Agent with caching enabled
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Cache agent decisions
from langchain.globals import set_llm_cache
from langchain.cache import RedisCache
set_llm_cache(RedisCache(redis_client))

agent = create_openai_functions_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

# Execute agent
result = agent_executor.invoke({"input": "What's the weather in San Francisco?"})
```

**Performance Optimizations:**
- Cache agent reasoning: 70% faster on repeated patterns
- Limit max_iterations: Prevent infinite loops
- Handle parsing errors: Graceful fallback
- Use gpt-4o-mini for simple tool selection

#### Streaming Agent Output
```python
# Stream agent steps
async def stream_agent():
    async for chunk in agent_executor.astream(
        {"input": "Research AI and summarize findings"}
    ):
        if "output" in chunk:
            print(chunk["output"], end="", flush=True)

asyncio.run(stream_agent())
```

**Benefits:**
- Real-time feedback on agent progress
- Better UX for long-running agents
- Visibility into reasoning steps

### 6. Memory Optimization (Context7-Verified)

**Pattern from Context7:**

#### Conversation Buffer Window Memory
```python
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.prompts import MessagesPlaceholder

# Keep only last 5 messages
memory = ConversationBufferWindowMemory(
    k=5,
    return_messages=True,
    memory_key="chat_history"
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{input}")
])

chain = (
    RunnablePassthrough.assign(
        chat_history=lambda x: memory.load_memory_variables({})["chat_history"]
    )
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# Use chain with memory
result = chain.invoke({"input": "Hello!"})
memory.save_context({"input": "Hello!"}, {"output": result})
```

**Benefits:**
- Prevents context overflow
- Maintains relevant history only
- 80% token savings vs full history
- No API errors from context limits

#### Summary Memory for Long Conversations
```python
from langchain.memory import ConversationSummaryMemory

# Automatically summarize old messages
memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    max_token_limit=2000
)

# Old messages are summarized
# Recent messages kept verbatim
# Total tokens stay under 2000
```

**Performance Impact:**
- Long conversations: 70% token reduction
- Cost savings: 70% fewer input tokens
- Quality maintained: Summary preserves context

### 7. Chain Composition Optimization (Context7-Verified)

**Pattern from Context7:**

#### Parallel Chain Execution
```python
from langchain_core.runnables import RunnableParallel

# Execute multiple chains in parallel
chain1 = ChatPromptTemplate.from_template("Summarize: {text}") | model
chain2 = ChatPromptTemplate.from_template("Extract keywords from: {text}") | model
chain3 = ChatPromptTemplate.from_template("Classify sentiment: {text}") | model

# Parallel execution
parallel_chain = RunnableParallel(
    summary=chain1,
    keywords=chain2,
    sentiment=chain3
)

result = parallel_chain.invoke({"text": "Article content here..."})
# Returns: {"summary": "...", "keywords": "...", "sentiment": "..."}

# Performance:
# Sequential: 3 × 2s = 6s
# Parallel: max(3 × 2s) = 2s (3x faster)
```

**Benefits:**
- Automatic parallel execution
- 3-5x faster for independent operations
- Simple composition syntax
- Built-in error handling

## Optimization Output

```
🔗 LangChain Application Optimization Analysis
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Project: LangChain Application
Current Usage: 500 requests/day
Monthly Cost: $450

📊 Current Performance Baseline
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  Chain Composition:
  - Manual chain construction
  - No LCEL patterns
  - Sequential execution only

  Caching:
  - No caching implemented
  - Duplicate queries: 40%

  RAG:
  - No embeddings cache
  - Basic similarity search
  - Linear vector search

  Async:
  - Sequential processing only
  - No batch operations

⚡ LCEL Pattern Optimization
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  Current: Manual chain construction
  Recommended: LCEL with | operator

  💡 Impact:
  - Automatic async/streaming support
  - Built-in retry and fallback logic
  - 3x faster parallel execution
  - Better debugging and monitoring

  LCEL chains configured ✓

🚀 Async and Streaming Optimization
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  ⚠️  Sequential processing only
  Current: 500 sequential requests

  💡 Recommendations:
  1. Use ainvoke/abatch for async processing → 5-10x faster
  2. Enable streaming → 10x better perceived latency
  3. Parallel chain execution → 3x speedup

  Async patterns configured ✓

  ⚡ Impact:
  - Sequential: 500 × 2s = 1,000s (~17 min)
  - Async batch: 500 / 50 concurrent × 2s = 20s (50x faster)
  - Time to first token: 2s → 500ms (4x faster)

💾 Caching Optimization
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  ⚠️  No caching, 40% duplicate queries
  Current: 200 duplicate queries/day

  💡 Recommendations:
  1. Redis cache → 2000x faster for cached queries
  2. Semantic cache → 60% cache hit rate
  3. Embeddings cache → 59x faster RAG

  Redis + semantic caching configured ✓

  ⚡ Impact:
  - Cached queries: 2s → <1ms (2000x faster)
  - Cache hit rate: 60% (300 queries/day)
  - Cost reduction: 60% fewer API calls
  - Monthly savings: $270

🔍 RAG Optimization
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  ⚠️  No embeddings cache, basic retrieval
  Issues: Slow queries, low diversity

  💡 Recommendations:
  1. Cache embeddings → 59x faster
  2. MMR retrieval → 40% better diversity
  3. Multi-query retrieval → 50% better coverage
  4. FAISS vector store → 50x faster search

  RAG optimizations configured ✓

  ⚡ Impact:
  - Embeddings: 1s → 17ms (59x faster)
  - Vector search: 500ms → 10ms (50x faster)
  - Retrieval quality: 60% → 85% (42% improvement)
  - Total RAG latency: 2s → 100ms (20x faster)

🎯 Summary
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  Total Optimizations: 18

  🔴 Critical: 5 (LCEL, async, caching, RAG, agents)
  🟡 High Impact: 8 (streaming, memory, parallel chains)
  🟢 Low Impact: 5 (monitoring, logging)

  Performance Improvements:

  Latency:
  - Async batch: 1,000s → 20s (50x faster)
  - Cached queries: 2s → <1ms (2000x faster)
  - RAG pipeline: 2s → 100ms (20x faster)
  - Time to first token: 2s → 500ms (4x faster)

  Cost Savings:
  - Caching: $270/month (60% reduction)
  - Embeddings cache: $180/month (40% reduction)
  - Total monthly savings: $450/month (100% reduction)
  - New monthly cost: $0 (cache only)

  Quality:
  - RAG relevance: 60% → 85% (42% improvement)
  - Retrieval diversity: +40%
  - Cache hit rate: 60%

  Run with --apply to implement optimizations
```

## Implementation

This command uses the **@langchain-expert** agent with optimization expertise:

1. Query Context7 for LangChain optimization patterns
2. Analyze current chain composition
3. Convert to LCEL patterns
4. Implement async and streaming
5. Configure caching strategies
6. Optimize RAG pipeline
7. Generate optimized code

## Best Practices Applied

Based on Context7 documentation from `/websites/python_langchain`:

1. **LCEL Composition** - 3x faster with automatic optimization
2. **Async Batch Processing** - 50x faster for multiple requests
3. **Semantic Caching** - 60% cache hit rate, 2000x faster
4. **Embeddings Caching** - 59x faster RAG queries
5. **MMR Retrieval** - 40% better diversity
6. **Streaming** - 4x better perceived latency
7. **Parallel Chains** - 3x faster for independent operations

## Related Commands

- `/rag:optimize` - RAG system optimization
- `/openai:optimize` - OpenAI API optimization
- `/anthropic:optimize` - Anthropic OpenCode optimization

## Troubleshooting

### Slow Chain Execution
- Convert to LCEL patterns (3x speedup)
- Enable async batch processing (50x speedup)
- Use streaming for long responses

### High Costs
- Enable Redis caching (60% reduction)
- Use semantic cache for query variations
- Cache embeddings for RAG (59x faster)

### Poor RAG Quality
- Use MMR retrieval for diversity
- Implement multi-query retrieval
- Optimize chunk size (1000 chars)
- Add 20% chunk overlap

### Memory Issues
- Use ConversationBufferWindowMemory
- Implement summary memory for long chats
- Limit max_tokens per request

## Installation

```bash
# Install LangChain
pip install langchain langchain-openai langchain-community

# Install caching support
pip install redis

# Install vector stores
pip install faiss-cpu

# Install async support
pip install aiohttp asyncio
```

## Version History

- v2.0.0 - Initial Schema v2.0 release with Context7 integration
- LCEL pattern optimization for 3x speedup
- Async batch processing for 50x throughput
- Redis semantic caching for 60% hit rate
- RAG optimizations (embeddings cache, MMR, multi-query)
- Streaming and parallel chain execution
- Memory optimization patterns