langchain

LangChain

LangChain is a composable framework for building applications with LLMs, featuring LangGraph for constructing robust, stateful, cyclic multi-agent architectures.

When to Use

Agentic Workflows & Multi-Turn Problem Solving: Building autonomous AI agents with LangGraph and tool calling.
Complex Retrieval-Augmented Generation (RAG): Orchestrating document chunking, embeddings, vector indexing, and reranking.
Declarative Chain Composition with LCEL: Composing streaming, async, and batchable chains via LangChain Expression Language.
Enterprise LLM Observability: Tracking token costs, latency, and step-by-step traces with LangSmith.

Quick Start

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

Declarative LCEL Pipe Syntax

prompt = ChatPromptTemplate.from_template("Summarize this concept in 3 bullet points: {topic}") model = ChatOpenAI(model="gpt-4o-mini") parser = StrOutputParser()

chain = prompt | model | parser result = chain.invoke({"topic": "Distributed Consensus"}) print(result)


## Core Concepts

### LangChain Expression Language (LCEL) & Streaming

Declarative pipe-based chain execution with type inference and streaming:

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
import os

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert cloud architect. Answer concisely."),
    ("user", "Explain the difference between {topic_a} and {topic_b}.")
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
parser = StrOutputParser()

# LCEL Pipeline: prompt | model | parser
chain = prompt | model | parser

# Stream output tokens in real-time
for chunk in chain.stream({"topic_a": "Kubernetes", "topic_b": "Docker Swarm"}):
    print(chunk, end="", flush=True)

Retrieval-Augmented Generation (RAG) Pipeline

Querying vector stores and passing relevant context to LLMs:

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI

# Mock vectorstore retriever
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["Antigravity IDE is an AI-first coding assistant.", "LangChain v0.3 introduces streamlined core abstractions."],
    embedding=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

rag_prompt = ChatPromptTemplate.from_template(
    "Answer the question based only on context: {context}\nQuestion: {question}"
)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

response = rag_chain.invoke("What does LangChain v0.3 introduce?")
print(response)

Agentic Workflows with Tool Calling

Binding custom Python functions as agent tools:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def calculate_shipping(weight_kg: float, destination: str) -> float:
    '''Calculate the shipping fee in USD for a parcel weight and destination.'''
    rate = 12.5 if destination.upper() == "US" else 28.0
    return round(weight_kg * rate, 2)

tools = [calculate_shipping]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

result = llm.invoke("What is the shipping cost for a 4.5kg package to the US?")
print("Tool Calls:", result.tool_calls)

Common Patterns

Stateful Agent Workflow with LangGraph

Problem: Legacy AgentExecutor loops lack checkpointing and cannot recover from partial errors.

Solution: Use LangGraph state graphs with message nodes and conditional edges:

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

def call_model(state: AgentState):
    # Process messages with LLM
    return {"messages": [model.invoke(state["messages"])]}

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.set_entry_point("agent")
workflow.add_edge("agent", END)
app = workflow.compile()

Best Practices

Do:

Target LangChain v0.3+ and construct pipelines using LCEL (chain = prompt | model | parser).
Use LangGraph for multi-turn, stateful, cyclical agent architectures rather than legacy AgentExecutor.
Configure LangSmith (LANGCHAIN_TRACING_V2=true) to monitor latency, cost, and tool calls in development and production.
Bind tools explicitly with Pydantic type annotations and docstrings for reliable tool call generation.

Don't:

Use deprecated v0.1 import paths (langchain.chains or langchain.agents); import from langchain_core and langgraph.
Construct prompts using manual string formatting; use ChatPromptTemplate to properly handle roles and escapes.
Pass unvalidated user input directly to vectorstore retriever filters.

Troubleshooting

Error	Cause	Solution
`OutputParserException: Failed to parse`	LLM response did not adhere to expected JSON schema format.	Add retry parser `OutputFixingParser` or use native model structured outputs.
`Recursion limit reached in LangGraph`	Infinite loop in graph conditional edge routing.	Inspect routing logic and increase `recursion_limit` in config if expected.
`RateLimitError in streaming chains`	Too many simultaneous tokens per second requested across chains.	Implement async concurrency throttler or use batching with `.batch()`.

References

LangChain Documentation