🇺🇸English

LangChain Orchestration Skill

Complete guide for building production-grade LLM applications with LangChain, covering chains, agents, memory, RAG patterns, and advanced orchestration techniques.

Table of Contents

1. Core Concepts
2. Chains
3. Agents
4. Memory Systems
5. RAG Patterns
6. LLM Integrations
7. Callbacks & Monitoring
8. Retrieval Strategies
9. Streaming
10. Error Handling
11. Production Best Practices

Core Concepts

LangChain Expression Language (LCEL)

LCEL is the declarative way to compose chains in LangChain, enabling streaming, async, and parallel execution.

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Basic LCEL chain
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
output_parser = StrOutputParser()

chain = prompt | llm | output_parser
result = chain.invoke({"topic": "quantum computing"})

Runnable Interface

Every component in LangChain implements the Runnable interface with standard methods:

from langchain_core.runnables import RunnablePassthrough

# Key methods: invoke, stream, batch, ainvoke, astream, abatch
chain = prompt | llm | output_parser

# Synchronous invoke
result = chain.invoke({"topic": "AI"})

# Streaming
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="", flush=True)

# Batch processing
results = chain.batch([{"topic": "AI"}, {"topic": "ML"}])

# Async variants
result = await chain.ainvoke({"topic": "AI"})

RunnablePassthrough

Pass inputs directly through or apply transformations:

from langchain_core.runnables import RunnablePassthrough

# Pass through unchanged
chain = RunnablePassthrough() | llm | output_parser

# With transformation
def add_context(x):
    return {"text": x["input"], "context": "important"}

chain = RunnablePassthrough.assign(processed=add_context) | llm

Chains

Sequential Chains

Process data through multiple steps sequentially.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)

# Step 1: Generate ideas
idea_prompt = ChatPromptTemplate.from_template(
    "Generate 3 creative ideas for: {topic}"
)
idea_chain = idea_prompt | llm | StrOutputParser()

# Step 2: Evaluate ideas
eval_prompt = ChatPromptTemplate.from_template(
    "Evaluate these ideas and pick the best one:\n{ideas}"
)
eval_chain = eval_prompt | llm | StrOutputParser()

# Combine into sequential chain
sequential_chain = (
    {"ideas": idea_chain}
    | RunnablePassthrough.assign(evaluation=eval_chain)
)

result = sequential_chain.invoke({"topic": "mobile app"})

Map-Reduce Chains

Process multiple inputs in parallel and combine results.

from langchain_core.runnables import RunnableParallel
from langchain_core.prompts import ChatPromptTemplate

# Define parallel processing
summary_prompt = ChatPromptTemplate.from_template(
    "Summarize this text in one sentence: {text}"
)
keywords_prompt = ChatPromptTemplate.from_template(
    "Extract 3 keywords from: {text}"
)
sentiment_prompt = ChatPromptTemplate.from_template(
    "Analyze sentiment (positive/negative/neutral): {text}"
)

# Map: Process in parallel
map_chain = RunnableParallel(
    summary=summary_prompt | llm | StrOutputParser(),
    keywords=keywords_prompt | llm | StrOutputParser(),
    sentiment=sentiment_prompt | llm | StrOutputParser()
)

# Reduce: Combine results
reduce_prompt = ChatPromptTemplate.from_template(
    """Combine the analysis:
    Summary: {summary}
    Keywords: {keywords}
    Sentiment: {sentiment}

    Provide a comprehensive report:"""
)

map_reduce_chain = map_chain | reduce_prompt | llm | StrOutputParser()

result = map_reduce_chain.invoke({
    "text": "LangChain is an amazing framework for building LLM applications."
})

Router Chains

Route inputs to different chains based on conditions.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Define specialized chains
technical_prompt = ChatPromptTemplate.from_template(
    "Provide a technical explanation of: {query}"
)
simple_prompt = ChatPromptTemplate.from_template(
    "Explain in simple terms: {query}"
)

technical_chain = technical_prompt | llm | StrOutputParser()
simple_chain = simple_prompt | llm | StrOutputParser()

# Router function
def route_query(input_dict):
    query = input_dict["query"]
    complexity = input_dict.get("complexity", "simple")

    if complexity == "technical":
        return technical_chain
    return simple_chain

# Create router chain
from langchain_core.runnables import RunnableLambda

router_chain = RunnableLambda(route_query)

# Use the router
result = router_chain.invoke({
    "query": "quantum entanglement",
    "complexity": "technical"
})

Conditional Chains

Execute chains based on conditions.

from langchain_core.runnables import RunnableBranch

# Define condition-based routing
classification_prompt = ChatPromptTemplate.from_template(
    "Classify this as 'question', 'statement', or 'command': {text}"
)

question_handler = ChatPromptTemplate.from_template(
    "Answer this question: {text}"
) | llm | StrOutputParser()

statement_handler = ChatPromptTemplate.from_template(
    "Acknowledge this statement: {text}"
) | llm | StrOutputParser()

command_handler = ChatPromptTemplate.from_template(
    "Execute this command: {text}"
) | llm | StrOutputParser()

# Create conditional branch
branch = RunnableBranch(
    (lambda x: "question" in x["type"].lower(), question_handler),
    (lambda x: "statement" in x["type"].lower(), statement_handler),
    command_handler  # default
)

# Full chain with classification
full_chain = (
    {"text": RunnablePassthrough(), "type": classification_prompt | llm | StrOutputParser()}
    | branch
)

LLMChain (Legacy)

Traditional chain format still supported:

from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    input_variables=["product"],
    template="What is a good name for a company that makes {product}?"
)

chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(product="eco-friendly water bottles")

Stuff Documents Chain

Combine documents into a single context:

from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.documents import Document

prompt = ChatPromptTemplate.from_template(
    """Answer based on the following context:

<context>
{context}
</context>

Question: {input}"""
)

document_chain = create_stuff_documents_chain(llm, prompt)

docs = [
    Document(page_content="LangChain supports multiple LLM providers."),
    Document(page_content="Chains can be composed using LCEL.")
]

result = document_chain.invoke({
    "input": "What does LangChain support?",
    "context": docs
})

Agents

ReAct Agents

Reasoning and Acting agents that use tools iteratively.

from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.tools import Tool
from langchain import hub

# Define tools
def search_tool(query: str) -> str:
    """Search for information"""
    return f"Search results for: {query}"

def calculator_tool(expression: str) -> str:
    """Calculate mathematical expressions"""
    try:
        return str(eval(expression))
    except:
        return "Invalid expression"

tools = [
    Tool(
        name="Search",
        func=search_tool,
        description="Useful for searching information"
    ),
    Tool(
        name="Calculator",
        func=calculator_tool,
        description="Useful for math calculations"
    )
]

# Create ReAct agent
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

result = agent_executor.invoke({
    "input": "What is 25 * 4, and then search for that number's significance"
})

LangGraph ReAct Agent

Modern approach using LangGraph for better control:

from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver

@tool
def retrieve(query: str) -> str:
    """Retrieve relevant information from the knowledge base"""
    # Your retrieval logic here
    return f"Retrieved information for: {query}"

@tool
def analyze(text: str) -> str:
    """Analyze text and provide insights"""
    return f"Analysis of: {text}"

# Create agent with memory
memory = MemorySaver()
agent_executor = create_react_agent(
    llm,
    [retrieve, analyze],
    checkpointer=memory
)

# Use with configuration
config = {"configurable": {"thread_id": "abc123"}}
for chunk in agent_executor.stream(
    {"messages": [("user", "Find information about LangChain")]},
    config=config
):
    print(chunk)

Conversational ReAct Agent

Agent with built-in conversation memory:

from langchain.agents import create_conversational_retrieval_agent
from langchain_core.tools import Tool

tools = [
    Tool(
        name="Knowledge Base",
        func=lambda q: f"KB result: {q}",
        description="Search the knowledge base"
    )
]

conversational_agent = create_conversational_retrieval_agent(
    llm,
    tools,
    verbose=True
)

# Maintains conversation context
result1 = conversational_agent.invoke({
    "input": "What is LangChain?"
})
result2 = conversational_agent.invoke({
    "input": "Tell me more about its features"
})

Zero-Shot React Agent

Agent that works without examples:

from langchain.agents import AgentType, initialize_agent, load_tools

# Load pre-built tools
tools = load_tools(["serpapi", "llm-math"], llm=llm)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    max_iterations=3
)

result = agent.run(
    "What is the population of Tokyo and what is that number divided by 2?"
)

Structured Chat Agent

Agent that uses structured input/output:

from langchain.agents import create_structured_chat_agent

# Define tools with structured schemas
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="The search query")
    max_results: int = Field(default=5, description="Maximum results")

@tool(args_schema=SearchInput)
def structured_search(query: str, max_results: int = 5) -> str:
    """Search with structured parameters"""
    return f"Found {max_results} results for: {query}"

tools = [structured_search]

prompt = hub.pull("hwchase17/structured-chat-agent")
agent = create_structured_chat_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Tool Calling Agent

Modern agent using native tool calling:

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the database"""
    return f"Found {limit} results for {query}"

# Bind tools to LLM
llm_with_tools = llm.bind_tools([multiply, search_database])

# Create simple tool chain
from operator import itemgetter

tool_chain = llm_with_tools | (lambda x: x.tool_calls[0]["args"]) | multiply
result = tool_chain.invoke("What's four times 23")

Memory Systems

ConversationBufferMemory

Store complete conversation history:

from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

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

chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

# Conversation is automatically stored
response1 = chain.run(input="Hi, I'm Alice")
response2 = chain.run(input="What's my name?")  # Will remember Alice

ConversationBufferWindowMemory

Keep only recent K interactions:

from langchain.memory import ConversationBufferWindowMemory

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

chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

ConversationSummaryMemory

Summarize conversation history:

from langchain.memory import ConversationSummaryMemory

memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

# Long conversations are automatically summarized
for i in range(20):
    chain.run(input=f"Tell me fact {i} about AI")

ConversationSummaryBufferMemory

Hybrid approach: recent messages + summary:

from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=100,  # When to trigger summarization
    memory_key="chat_history",
    return_messages=True
)

Vector Store Memory

Semantic search over conversation history:

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts([], embeddings)

memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# Save context
memory.save_context(
    {"input": "My favorite color is blue"},
    {"output": "That's great!"}
)

# Retrieve relevant context
relevant = memory.load_memory_variables({"input": "What's my favorite color?"})

Recall Memories (LangGraph)

Structured memory with save and search:

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.tools import tool

recall_vector_store = InMemoryVectorStore(OpenAIEmbeddings())

@tool
def save_recall_memory(memory: str) -> str:
    """Save important information to long-term memory"""
    recall_vector_store.add_texts([memory])
    return f"Saved memory: {memory}"

@tool
def search_recall_memories(query: str) -> str:
    """Search long-term memories"""
    docs = recall_vector_store.similarity_search(query, k=3)
    return "\n".join([doc.page_content for doc in docs])

# Use with agent
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    llm,
    [save_recall_memory, search_recall_memories]
)

Custom Memory with LangGraph State

Define custom state for memory:

from typing import List
from langgraph.graph import MessagesState, StateGraph, START, END

class State(MessagesState):
    recall_memories: List[str]

def load_memories(state: State):
    """Load relevant memories before agent processes input"""
    messages = state["messages"]
    last_message = messages[-1].content if messages else ""

    # Search for relevant memories
    docs = recall_vector_store.similarity_search(last_message, k=3)
    memories = [doc.page_content for doc in docs]

    return {"recall_memories": memories}

# Add to graph
builder = StateGraph(State)
builder.add_node(load_memories)
builder.add_edge(START, "load_memories")

RAG Patterns

Basic RAG Chain

Fundamental retrieval-augmented generation:

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# Setup vector store
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    [
        "LangChain supports multiple LLM providers including OpenAI, Anthropic, and more.",
        "Chains can be composed using LangChain Expression Language (LCEL).",
        "Agents can use tools to interact with external systems."
    ],
    embedding=embeddings
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

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

{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Build RAG chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

result = rag_chain.invoke("What does LangChain support?")

RAG with Retrieval Chain

Using built-in retrieval chain constructor:

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

prompt = ChatPromptTemplate.from_template(
    """Answer based on the context:

<context>
{context}
</context>

Question: {input}"""
)

document_chain = create_stuff_documents_chain(llm, prompt)
retrieval_chain = create_retrieval_chain(retriever, document_chain)

response = retrieval_chain.invoke({
    "input": "What is LCEL?"
})
# Returns: {"input": "...", "context": [...], "answer": "..."}

RAG with Chat History

Conversational RAG with context:

from langchain.chains import create_history_aware_retriever
from langchain_core.prompts import MessagesPlaceholder

contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Given a chat history and the latest user question, "
               "formulate a standalone question which can be understood "
               "without the chat history."),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

history_aware_retriever = create_history_aware_retriever(
    llm,
    retriever,
    contextualize_prompt
)

# Use in RAG chain
qa_chain = create_retrieval_chain(
    history_aware_retriever,
    document_chain
)

# First question
result1 = qa_chain.invoke({
    "input": "What is LangChain?",
    "chat_history": []
})

# Follow-up with context
result2 = qa_chain.invoke({
    "input": "What are its main features?",
    "chat_history": [
        ("human", "What is LangChain?"),
        ("ai", result1["answer"])
    ]
})

Multi-Query RAG

Generate multiple search queries for better retrieval:

from langchain.retrievers.multi_query import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

# Automatically generates multiple query variations
rag_chain = (
    {"context": multi_query_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

RAG with Reranking

Improve relevance with reranking:

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import FlashrankRerank

# Setup reranker
compressor = FlashrankRerank()
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

# Use in RAG chain
rag_chain = (
    {"context": compression_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

Parent Document Retrieval

Retrieve larger parent documents for full context:

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Storage for parent documents
store = InMemoryStore()

# Splitters
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# Add documents
parent_retriever.add_documents(documents)

Self-Query Retrieval

Natural language to structured queries:

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

metadata_field_info = [
    AttributeInfo(
        name="source",
        description="The document source",
        type="string",
    ),
    AttributeInfo(
        name="page",
        description="The page number",
        type="integer",
    ),
]

document_content_description = "Technical documentation"

self_query_retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    document_content_description,
    metadata_field_info,
)

LLM Integrations

OpenAI Integration

from langchain_openai import ChatOpenAI, OpenAI

# Chat model
chat_model = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,
    max_tokens=500,
    api_key="your-api-key"
)

# Completion model
completion_model = OpenAI(
    model="gpt-3.5-turbo-instruct",
    temperature=0.9
)

Anthropic Claude Integration

from langchain_anthropic import ChatAnthropic

claude = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
    max_tokens=1024,
    api_key="your-api-key"
)

HuggingFace Integration

from langchain_huggingface import HuggingFaceEndpoint

llm = HuggingFaceEndpoint(
    repo_id="meta-llama/Llama-2-7b-chat-hf",
    huggingfacehub_api_token="your-token",
    task="text-generation",
    temperature=0.7
)

Google Vertex AI Integration

from langchain_google_vertexai import ChatVertexAI, VertexAI

# Chat model
chat_model = ChatVertexAI(
    model_name="chat-bison",
    temperature=0
)

# Completion model
completion_model = VertexAI(
    model_name="gemini-1.0-pro-002"
)

Ollama Local Models

from langchain_community.llms import Ollama

llm = Ollama(
    model="llama2",
    temperature=0.8
)

Binding Tools to LLMs

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers together"""
    return a * b

# Bind tools to model
llm_with_tools = llm.bind_tools([multiply])

# Model will return tool calls
response = llm_with_tools.invoke("What is 3 times 4?")
print(response.tool_calls)

Callbacks & Monitoring

Standard Callbacks

Track chain execution:

from langchain_core.callbacks import StdOutCallbackHandler
from langchain.callbacks import get_openai_callback

# Standard output callback
callbacks = [StdOutCallbackHandler()]

chain = prompt | llm | StrOutputParser()
result = chain.invoke(
    {"topic": "AI"},
    config={"callbacks": callbacks}
)

# OpenAI cost tracking
with get_openai_callback() as cb:
    result = chain.invoke({"topic": "AI"})
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Total Cost: ${cb.total_cost}")

Custom Callbacks

Create custom callback handlers:

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

class MyCustomCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized: Dict[str, Any], prompts: list[str], **kwargs):
        print(f"LLM started with prompts: {prompts}")

    def on_llm_end(self, response, **kwargs):
        print(f"LLM finished with response: {response}")

    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs):
        print(f"Chain started with inputs: {inputs}")

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs):
        print(f"Chain ended with outputs: {outputs}")

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs):
        print(f"Tool started with input: {input_str}")

    def on_tool_end(self, output: str, **kwargs):
        print(f"Tool ended with output: {output}")

# Use custom callback
custom_callback = MyCustomCallback()
result = chain.invoke(
    {"topic": "AI"},
    config={"callbacks": [custom_callback]}
)

Argilla Callback

Track and log to Argilla:

from langchain_community.callbacks import ArgillaCallbackHandler

argilla_callback = ArgillaCallbackHandler(
    dataset_name="langchain-dataset",
    api_url="http://localhost:6900",
    api_key="your-api-key"
)

callbacks = [argilla_callback]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    callbacks=callbacks
)

agent.run("Who was the first president of the United States?")

UpTrain Callback

RAG evaluation and monitoring:

from langchain_community.callbacks import UpTrainCallbackHandler

uptrain_callback = UpTrainCallbackHandler(
    key_type="uptrain",
    api_key="your-api-key"
)

config = {"callbacks": [uptrain_callback]}

# Automatically evaluates context relevance, factual accuracy, completeness
result = rag_chain.invoke("What is LangChain?", config=config)

LangSmith Integration

Production monitoring and debugging:

import os

# Set environment variables
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

# All chains automatically traced
result = chain.invoke({"topic": "AI"})
# View traces at smith.langchain.com

Retrieval Strategies

Vector Store Retrievers

Basic similarity search:

from langchain_community.vectorstores import FAISS, Chroma, Pinecone

# FAISS
faiss_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# Maximum Marginal Relevance (MMR)
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5}
)

# Similarity with threshold
threshold_retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8, "k": 5}
)

Ensemble Retriever

Combine multiple retrievers:

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 for keyword search
bm25_retriever = BM25Retriever.from_texts(texts)
bm25_retriever.k = 5

# Combine with vector search
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, faiss_retriever],
    weights=[0.5, 0.5]
)

docs = ensemble_retriever.get_relevant_documents("LangChain features")

Time-Weighted Retriever

Prioritize recent documents:

from langchain.retrievers import TimeWeightedVectorStoreRetriever

retriever = TimeWeightedVectorStoreRetriever(
    vectorstore=vectorstore,
    decay_rate=0.01,  # Decay factor for older docs
    k=5
)

Multi-Vector Retriever

Multiple vectors per document:

from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryByteStore

store = InMemoryByteStore()

retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key="doc_id"
)

# Add documents with multiple representations
retriever.add_documents(documents)

Streaming

Stream Chain Output

Stream tokens as they're generated:

from langchain_core.output_parsers import StrOutputParser

chain = prompt | llm | StrOutputParser()

# Stream method
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="", flush=True)

Stream with Callbacks

Handle streaming events:

from langchain_core.callbacks import StreamingStdOutCallbackHandler

streaming_llm = ChatOpenAI(
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

chain = prompt | streaming_llm | StrOutputParser()
result = chain.invoke({"topic": "AI"})  # Streams to stdout

Async Streaming

Stream asynchronously:

async def stream_async():
    async for chunk in chain.astream({"topic": "AI"}):
        print(chunk, end="", flush=True)

# Run async
import asyncio
asyncio.run(stream_async())

Stream Agent Responses

Stream agent execution:

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools)

for chunk in agent.stream(
    {"messages": [("user", "Search for LangChain information")]},
    stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

Streaming RAG

Stream RAG responses:

retrieval_chain = (
    {
        "context": retriever.with_config(run_name="Docs"),
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

# Stream the response
for chunk in retrieval_chain.stream("What is LangChain?"):
    print(chunk, end="", flush=True)

Error Handling

Retry Logic

Automatic retries on failure:

from langchain_core.runnables import RunnableRetry

# Add retry to chain
chain_with_retry = (prompt | llm | StrOutputParser()).with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

result = chain_with_retry.invoke({"topic": "AI"})

Fallback Chains

Use fallback on errors:

from langchain_core.runnables import RunnableWithFallbacks

primary_llm = ChatOpenAI(model="gpt-4")
fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")

chain_with_fallback = (prompt | primary_llm).with_fallbacks(
    [prompt | fallback_llm]
)

result = chain_with_fallback.invoke({"topic": "AI"})

Try-Except Patterns

Manual error handling:

from langchain_core.exceptions import OutputParserException

try:
    result = chain.invoke({"topic": "AI"})
except OutputParserException as e:
    print(f"Parsing failed: {e}")
    result = chain.invoke({"topic": "AI"})  # Retry
except Exception as e:
    print(f"Chain execution failed: {e}")
    result = None

Timeout Handling

Set execution timeouts:

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(timeout=10.0)  # 10 seconds

try:
    result = chain.invoke({"topic": "AI"}, config=config)
except TimeoutError:
    print("Chain execution timed out")

Validation

Validate inputs and outputs:

from pydantic import BaseModel, Field, validator

class QueryInput(BaseModel):
    topic: str = Field(..., min_length=1, max_length=100)

    @validator("topic")
    def topic_must_be_valid(cls, v):
        if not v.strip():
            raise ValueError("Topic cannot be empty")
        return v.strip()

# Use with chain
def validate_and_invoke(topic: str):
    try:
        validated = QueryInput(topic=topic)
        return chain.invoke({"topic": validated.topic})
    except ValueError as e:
        return f"Validation error: {e}"

Production Best Practices

Environment Configuration

Manage secrets securely:

import os
from dotenv import load_dotenv

load_dotenv()

# Use environment variables
llm = ChatOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    model=os.getenv("MODEL_NAME", "gpt-4o-mini")
)

# Vector store configuration
VECTOR_STORE_TYPE = os.getenv("VECTOR_STORE", "faiss")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")

Caching

Cache LLM responses:

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

# In-memory cache
set_llm_cache(InMemoryCache())

# Persistent cache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# Responses are cached automatically
result1 = llm.invoke("What is AI?")  # Calls API
result2 = llm.invoke("What is AI?")  # Uses cache

Rate Limiting

Control API usage:

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=1,
    check_every_n_seconds=0.1,
    max_bucket_size=10
)

llm = ChatOpenAI(rate_limiter=rate_limiter)

Batch Processing

Process multiple inputs efficiently:

# Batch invoke
inputs = [{"topic": f"Topic {i}"} for i in range(10)]
results = chain.batch(inputs, config={"max_concurrency": 5})

# Async batch
async def batch_process():
    results = await chain.abatch(inputs)
    return results

Monitoring and Logging

Production monitoring:

import logging
from langchain_core.callbacks import BaseCallbackHandler

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionCallback(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        logger.info(f"Chain started: {serialized.get('name', 'unknown')}")

    def on_chain_end(self, outputs, **kwargs):
        logger.info(f"Chain completed successfully")

    def on_chain_error(self, error, **kwargs):
        logger.error(f"Chain error: {error}")

# Use in production
production_callback = ProductionCallback()
config = {"callbacks": [production_callback]}

Testing Chains

Unit test your chains:

import pytest
from langchain_core.messages import HumanMessage, AIMessage

def test_basic_chain():
    chain = prompt | llm | StrOutputParser()
    result = chain.invoke({"topic": "testing"})
    assert isinstance(result, str)
    assert len(result) > 0

def test_rag_chain():
    result = rag_chain.invoke("What is LangChain?")
    assert "LangChain" in result
    assert len(result) > 50

@pytest.mark.asyncio
async def test_async_chain():
    result = await chain.ainvoke({"topic": "async"})
    assert isinstance(result, str)

Performance Optimization

Optimize chain execution:

# Use appropriate chunk sizes for text splitting
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)

# Limit retrieval results
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# Use smaller, faster models where appropriate
fast_llm = ChatOpenAI(model="gpt-4o-mini")

# Enable streaming for better UX
streaming_chain = prompt | fast_llm.with_streaming() | StrOutputParser()

Documentation

Document your chains:

from langchain_core.runnables import RunnableConfig

class DocumentedChain:
    """
    Production RAG chain for technical documentation.

    Features:
    - Multi-query retrieval for better coverage
    - Reranking for improved relevance
    - Streaming support
    - Error handling with fallbacks

    Usage:
        chain = DocumentedChain()
        result = chain.invoke("Your question here")
    """

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        self.retriever = self._setup_retriever()
        self.chain = self._build_chain()

    def _setup_retriever(self):
        # Setup logic
        pass

    def _build_chain(self):
        # Chain construction
        pass

    def invoke(self, query: str, config: RunnableConfig = None):
        """Execute the chain with error handling"""
        try:
            return self.chain.invoke(query, config=config)
        except Exception as e:
            logger.error(f"Chain execution failed: {e}")
            raise

---

Summary

This skill covers comprehensive LangChain orchestration patterns:

• Chains : Sequential, map-reduce, router, conditional chains
• Agents : ReAct, conversational, zero-shot, structured agents
• Memory : Buffer, window, summary, vector store memory
• RAG : Basic, multi-query, reranking, parent document retrieval
• LLM Integration : OpenAI, Anthropic, HuggingFace, Vertex AI, Ollama
• Callbacks : Standard, custom, Argilla, UpTrain, LangSmith
• Retrieval : Vector store, ensemble, time-weighted, multi-vector
• Streaming : Chain, agent, async streaming
• Error Handling : Retry, fallback, timeout, validation
• Production : Configuration, caching, rate limiting, monitoring, testing

For more examples and patterns, see EXAMPLES.md.

Weekly Installs

62

Repository

manutej/luxor-c…ketplace

GitHub Stars

47

First Seen

Jan 22, 2026

Security Audits

Gen Agent Trust HubFailSocketPassSnykWarn

Installed on

opencode48

codex47

gemini-cli46

github-copilot44

cursor43

claude-code41