|$ curl https://forge-ai.dev/api/markdown?path=docs/ai/langchain

$cat docs/langchain-framework.md

updated Recently·40 min read·published

LangChain Framework

◆AI◆Intermediate to Advanced🎯Free Tools

Introduction

LangChain is an open-source framework for building applications powered by large language models. It provides composable abstractions for chaining together LLM calls, retrieval systems, tools, and memory — enabling developers to build complex AI workflows from simple, reusable components.

The framework's core philosophy is modularity: each component (model, prompt, retriever, tool) is a building block that can be combined using LangChain Expression Language (LCEL) to create sophisticated pipelines. This guide covers the essential building blocks and patterns for production applications.

Installation & Setup

LangChain is split into multiple packages. The core package provides base abstractions, while partner packages add specific integrations.

install.sh

Bash

1	# Core LangChain
2	pip install langchain
3
4	# Partner packages (install what you need)
5	pip install langchain-openai
6	pip install langchain-anthropic
7	pip install langchain-community
8	pip install langchain-core
9
10	# Vector stores
11	pip install langchain-pinecone langchain-chroma
12
13	# Loaders for document ingestion
14	pip install langchain-community[document_loaders]

setup.py

Python

1	import os
2	from langchain_openai import ChatOpenAI
3	from langchain_core.messages import HumanMessage
4
5	os.environ["OPENAI_API_KEY"] = "sk-..."
6
7	llm = ChatOpenAI(model="gpt-4o", temperature=0)
8	response = llm.invoke([HumanMessage(content="What is LangChain?")])
9	print(response.content)

ℹ

info

Always set API keys via environment variables, never hardcode them. Use .env files locally and secrets managers in production.

LangChain Expression Language (LCEL)

LCEL is the declarative way to compose LangChain components using the pipe (|) operator. It automatically handles streaming, async, batch operations, and fallbacks.

lcel_basics.py

Python

1	from langchain_openai import ChatOpenAI
2	from langchain_core.prompts import ChatPromptTemplate
3	from langchain_core.output_parsers import StrOutputParser
4
5	# Compose a chain using LCEL pipe operator
6	prompt = ChatPromptTemplate.from_messages([
7	("system", "You are a helpful assistant. Answer in {style}."),
8	("human", "{question}"),
9	])
10
11	llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
12	parser = StrOutputParser()
13
14	# Chain using pipe operator — automatically handles
15	# streaming, batching, async, and error propagation
16	chain = prompt \| llm \| parser
17
18	# Invoke the chain
19	result = chain.invoke({
20	"style": "concise",
21	"question": "What is the LangChain Expression Language?"
22	})
23	print(result)

lcel_parallel.py

Python

1	# Parallel branches using RunnableParallel
2	from langchain_core.runnables import RunnableParallel
3
4	# Run two chains in parallel, merge results
5	analysis_chain = prompt \| llm \| parser
6
7	parallel = RunnableParallel(
8	summary=analysis_chain,
9	keywords=ChatPromptTemplate.from_template(
10	"List 5 keywords for: {question}"
11	) \| llm \| parser,
12	)
13
14	result = parallel.invoke({"style": "brief", "question": "Explain RAG"})
15	# result["summary"] and result["keywords"] both populated
16	print(result)

lcel_streaming.py

Python

1	# LCEL with streaming support
2	async for chunk in chain.astream({
3	"style": "concise",
4	"question": "What is retrieval-augmented generation?"
5	}):
6	print(chunk, end="", flush=True)
7
8	# Batch processing — run multiple inputs
9	results = chain.batch([
10	{"style": "concise", "question": "What is RAG?"},
11	{"style": "detailed", "question": "What is RAG?"},
12	{"style": "bullet-points", "question": "What is RAG?"},
13	])

✓

best practice

Always prefer LCEL over the legacy LLMChain class. LCEL provides native streaming, async support, and better error handling with less code.

Chains

Chains are sequences of calls that pass data from one component to the next. Common patterns include sequential chains, router chains, and retrieval chains.

sequential_chain.py

Python

1	# Sequential chain — output of one step feeds the next
2	from langchain_openai import ChatOpenAI
3	from langchain_core.prompts import ChatPromptTemplate
4	from langchain_core.output_parsers import StrOutputParser
5
6	llm = ChatOpenAI(model="gpt-4o", temperature=0)
7
8	# Step 1: Generate an outline
9	outline_prompt = ChatPromptTemplate.from_template(
10	"Create a brief outline for a blog post about: {topic}"
11	)
12	outline_chain = outline_prompt \| llm \| StrOutputParser()
13
14	# Step 2: Expand the outline into full content
15	content_prompt = ChatPromptTemplate.from_template(
16	"Write a blog post based on this outline:\n{outline}"
17	)
18	content_chain = content_prompt \| llm \| StrOutputParser()
19
20	# Compose: topic -> outline -> content
21	full_chain = (
22	{"outline": outline_chain}
23	\| (lambda x: {"outline": x["outline"]})
24	\| content_chain
25	)
26
27	result = full_chain.invoke({"topic": "Building AI Agents with LangChain"})
28	print(result)

router_chain.py

Python

1	# Router chain — dynamically choose a processing path
2	from langchain_core.runnables import RunnableLambda
3
4	def route_to_specialist(input_dict):
5	topic = input_dict["topic"].lower()
6	if "python" in topic or "code" in topic:
7	return python_specialist
8	elif "math" in topic or "equation" in topic:
9	return math_specialist
10	return general_specialist
11
12	python_specialist = (
13	ChatPromptTemplate.from_template(
14	"You are a Python expert. Answer: {question}"
15	) \| llm \| StrOutputParser()
16	)
17
18	math_specialist = (
19	ChatPromptTemplate.from_template(
20	"You are a math expert. Show your work for: {question}"
21	) \| llm \| StrOutputParser()
22	)
23
24	general_specialist = (
25	ChatPromptTemplate.from_template(
26	"Answer this question clearly: {question}"
27	) \| llm \| StrOutputParser()
28	)
29
30	router = RunnableLambda(route_to_specialist)
31	result = router.invoke({"topic": "python async", "question": "Explain async/await"})

ℹ

info

Use RunnableLambda for custom routing logic. For more complex routing with structured output, consider using LangGraph which provides a graph-based approach to agent workflows.

Output Parsers

Output parsers transform raw LLM text into structured data. LangChain provides parsers for strings, JSON, Pydantic models, and custom schemas — essential for building reliable pipelines that consume LLM output programmatically.

str_parser.py

Python

1	# StrOutputParser — simplest parser, returns raw string
2	from langchain_openai import ChatOpenAI
3	from langchain_core.prompts import ChatPromptTemplate
4	from langchain_core.output_parsers import StrOutputParser
5
6	llm = ChatOpenAI(model="gpt-4o", temperature=0)
7	prompt = ChatPromptTemplate.from_template(
8	"List 3 benefits of {topic}. One per line."
9	)
10
11	chain = prompt \| llm \| StrOutputParser()
12	result = chain.invoke({"topic": "Docker"})
13	# Returns a plain string, not an AIMessage object
14	print(type(result)) # <class 'str'>

json_parser.py

Python

1	# JsonOutputParser — parse LLM output into a dict
2	from langchain_core.output_parsers import JsonOutputParser
3	from langchain_core.prompts import ChatPromptTemplate
4
5	parser = JsonOutputParser()
6
7	prompt = ChatPromptTemplate.from_messages([
8	("system", "You are a helpful assistant.\n{format_instructions}"),
9	("human", "{query}"),
10	])
11
12	chain = prompt \| llm \| parser
13	result = chain.invoke({
14	"query": "Compare Python vs JavaScript for web backends",
15	"format_instructions": parser.get_format_instructions(),
16	})
17	# Returns a dict with the parsed JSON
18	print(result["comparison"])

pydantic_parser.py

Python

1	# PydanticOutputParser — type-safe structured output
2	from langchain_core.output_parsers import PydanticOutputParser
3	from pydantic import BaseModel, Field
4
5	class CodeReview(BaseModel):
6	summary: str = Field(description="One-sentence summary")
7	issues: list[str] = Field(description="List of issues found")
8	score: int = Field(description="Quality score 1-10")
9
10	parser = PydanticOutputParser(pydantic_object=CodeReview)
11
12	prompt = ChatPromptTemplate.from_messages([
13	("system", "Review code.\n{format_instructions}"),
14	("human", "{code}"),
15	])
16
17	chain = prompt \| llm \| parser
18	result = chain.invoke({
19	"code": "def add(a,b): return a+b",
20	"format_instructions": parser.get_format_instructions(),
21	})
22	# Returns a CodeReview instance with validated fields
23	print(result.score) # int
24	print(result.issues) # list[str]

ℹ

info

Start with StrOutputParser for simple use cases, then graduate to PydanticOutputParser when you need validated, typed data. Use with_structured_output() on the LLM directly for the cleanest approach — it handles parsing internally.

Memory

Memory components store conversation history and context between invocations. LangChain provides several memory strategies from simple buffer memory to sophisticated summarization.

memory_buffer.py

Python

1	# Conversation history with buffer memory
2	from langchain_openai import ChatOpenAI
3	from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
4	from langchain_core.messages import HumanMessage, AIMessage
5	from langchain_core.output_parsers import StrOutputParser
6
7	llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
8
9	# In-memory conversation store
10	chat_history = []
11
12	prompt = ChatPromptTemplate.from_messages([
13	("system", "You are a helpful coding assistant."),
14	MessagesPlaceholder(variable_name="history"),
15	("human", "{input}"),
16	])
17
18	chain = prompt \| llm \| StrOutputParser()
19
20	def chat(user_input: str) -> str:
21	response = chain.invoke({
22	"history": chat_history,
23	"input": user_input,
24	})
25	# Store the exchange in history
26	chat_history.append(HumanMessage(content=user_input))
27	chat_history.append(AIMessage(content=response))
28	return response
29
30	# Multi-turn conversation
31	print(chat("What is a closure in JavaScript?"))
32	print(chat("Show me an example")) # Context retained
33	print(chat("How is that different from a scope?")) # Full history used

memory_advanced.py

Python

1	# Summary memory — compress long conversations
2	from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
3
4	summary = ""
5
6	summarize_prompt = ChatPromptTemplate.from_template(
7	"Summarize this conversation so far:\n{history}\n"
8	"New message: {new_message}\n"
9	"Provide a concise summary."
10	)
11
12	def summarize_and_append(history, new_message):
13	global summary
14	result = (summarize_prompt \| llm \| StrOutputParser()).invoke({
15	"history": history,
16	"new_message": new_message,
17	})
18	summary = result
19	return summary
20
21	# RAG-backed memory — store in vector database
22	from langchain_chroma import Chroma
23	from langchain_openai import OpenAIEmbeddings
24
25	vectorstore = Chroma(
26	collection_name="chat_memory",
27	embedding_function=OpenAIEmbeddings(),
28	)
29
30	def remember(key: str, value: str):
31	vectorstore.add_texts([value], metadatas=[{"key": key}])
32
33	def recall(query: str, k: int = 3):
34	docs = vectorstore.similarity_search(query, k=k)
35	return [doc.page_content for doc in docs]

⚠

warning

Buffer memory grows unbounded — always implement a sliding window or summarization strategy for production use. Conversation history in memory is lost on restart unless persisted to a database.

Tools & Function Calling

Tools are functions that an LLM can call to interact with external systems. LangChain provides a decorator-based interface for defining tools with type-safe schemas.

tools_basics.py

Python

1	from langchain_core.tools import tool
2	from langchain_openai import ChatOpenAI
3	from langchain_core.messages import HumanMessage
4	import math
5
6	# Define tools using the @tool decorator
7	@tool
8	def calculate(expression: str) -> str:
9	"""Evaluate a mathematical expression. Supports basic arithmetic,
10	trigonometry, and common math functions."""
11	allowed = {
12	"sqrt": math.sqrt, "abs": abs, "round": round,
13	"sin": math.sin, "cos": math.cos, "pi": math.pi,
14	}
15	try:
16	result = eval(expression, {"__builtins__": {}}, allowed)
17	return str(result)
18	except Exception as e:
19	return f"Error: {e}"
20
21	@tool
22	def get_weather(city: str) -> str:
23	"""Get current weather for a given city."""
24	# In production, call a real weather API
25	weather_data = {
26	"New York": "72F, sunny",
27	"London": "58F, cloudy",
28	"Tokyo": "75F, clear",
29	}
30	return weather_data.get(city, f"Weather data unavailable for {city}")
31
32	# Bind tools to LLM
33	llm = ChatOpenAI(model="gpt-4o", temperature=0)
34	llm_with_tools = llm.bind_tools([calculate, get_weather])
35
36	# LLM decides which tool to call
37	response = llm_with_tools.invoke([
38	HumanMessage(content="What is sqrt(144) + cos(pi)?")
39	])
40	print(response.tool_calls)
41	# [{'name': 'calculate', 'args': {'expression': 'sqrt(144) + cos(pi)'}}]

agent_tools.py

Python

1	# Agent that uses tools autonomously
2	from langgraph.prebuilt import create_react_agent
3
4	# Create a ReAct agent with tools
5	agent = create_react_agent(
6	model=llm,
7	tools=[calculate, get_weather],
8	prompt="You are a helpful assistant. Use tools when needed.",
9	)
10
11	# Agent will reason, decide which tool to call, and respond
12	result = agent.invoke({
13	"messages": [HumanMessage(content="What is 42 * 3.14159 and what's the weather in Tokyo?")]
14	})
15
16	for msg in result["messages"]:
17	if hasattr(msg, "tool_calls") and msg.tool_calls:
18	for tc in msg.tool_calls:
19	print(f"Tool called: {tc['name']}({tc['args']})")
20	elif msg.content:
21	print(f"Response: {msg.content}")

✓

best practice

Always write clear, descriptive docstrings for your tools — the LLM uses them to decide when and how to call each tool. Test tools independently before binding them to agents.

Document Loaders

Document loaders ingest data from external sources — files, databases, APIs, web pages — into LangChain Document objects. Each loader handles parsing, metadata extraction, and chunking differently depending on the source format.

file_loaders.py

Python

1	# Text and file loaders
2	from langchain_community.document_loaders import (
3	TextLoader,
4	CSVLoader,
5	JSONLoader,
6	DirectoryLoader,
7	)
8
9	# Load a single text file
10	loader = TextLoader("README.md", encoding="utf-8")
11	docs = loader.load()
12	print(docs[0].metadata) # {'source': 'README.md'}
13
14	# Load CSV rows — each row becomes a Document
15	csv_loader = CSVLoader("data.csv", csv_args={"delimiter": ","})
16	csv_docs = csv_loader.load()
17	# csv_docs[0].page_content = "Name: Alice\nAge: 30\n..."
18
19	# Load all .md files from a directory
20	dir_loader = DirectoryLoader(
21	"./docs",
22	glob="*/.md",
23	loader_cls=TextLoader,
24	show_progress=True,
25	)
26	all_docs = dir_loader.load()
27	print(f"Loaded {len(all_docs)} documents")

web_loaders.py

Python

1	# Web and API loaders
2	from langchain_community.document_loaders import WebBaseLoader
3	from langchain_community.document_loaders import GitbookLoader
4
5	# Scrape a web page
6	web_loader = WebBaseLoader("https://example.com/article")
7	web_docs = web_loader.load()
8	print(web_docs[0].page_content[:200]) # Extracted text content
9
10	# Load from Gitbook or Notion
11	notion_loader = NotionDatabaseLoader(
12	integration_token="ntn_...",
13	database_id="abc123",
14	)
15	notion_docs = notion_loader.load()
16
17	# PDF loader — requires pip install pypdf
18	from langchain_community.document_loaders import PyPDFLoader
19	pdf_loader = PyPDFLoader("paper.pdf")
20	pages = pdf_loader.load() # One Document per page

ℹ

info

Use DirectoryLoader with glob patterns to load entire folders. For production RAG, combine loaders with RecursiveCharacterTextSplitter to chunk documents before embedding.

Retrieval Chains

Retrieval chains combine document retrieval with LLM generation — the foundation of RAG (Retrieval-Augmented Generation). LangChain simplifies document loading, splitting, embedding, and retrieval.

rag_chain.py

Python

1	# Complete RAG pipeline
2	from langchain_openai import ChatOpenAI, OpenAIEmbeddings
3	from langchain_chroma import Chroma
4	from langchain_core.prompts import ChatPromptTemplate
5	from langchain_core.output_parsers import StrOutputParser
6	from langchain_core.runnables import RunnablePassthrough
7
8	# 1. Load and split documents
9	from langchain_community.document_loaders import TextLoader
10	from langchain_text_splitters import RecursiveCharacterTextSplitter
11
12	loader = TextLoader("docs.txt")
13	docs = loader.load()
14
15	splitter = RecursiveCharacterTextSplitter(
16	chunk_size=1000,
17	chunk_overlap=200,
18	separators=["\n\n", "\n", ". ", " "],
19	)
20	splits = splitter.split_documents(docs)
21
22	# 2. Create vector store
23	vectorstore = Chroma.from_documents(
24	documents=splits,
25	embedding=OpenAIEmbeddings(),
26	persist_directory="./chroma_db",
27	)
28
29	# 3. Create retriever
30	retriever = vectorstore.as_retriever(
31	search_type="similarity",
32	search_kwargs={"k": 4},
33	)
34
35	# 4. Build RAG chain
36	rag_prompt = ChatPromptTemplate.from_template("""
37	Answer the question based on the context below.
38	If the answer is not in the context, say "I don't have enough information."
39
40	Context: {context}
41	Question: {question}
42	""")
43
44	def format_docs(docs):
45	return "\n\n".join(doc.page_content for doc in docs)
46
47	rag_chain = (
48	{"context": retriever \| format_docs, "question": RunnablePassthrough()}
49	\| rag_prompt
50	\| ChatOpenAI(model="gpt-4o")
51	\| StrOutputParser()
52	)
53
54	answer = rag_chain.invoke("What are the main features?")
55	print(answer)

ℹ

info

Use RunnablePassthrough to pass through the original input alongside transformed data. This pattern keeps your chain composable and testable.

Agents with LangGraph

LangGraph is the recommended way to build stateful, multi-step agents. It models agent workflows as graphs with nodes (operations) and edges (transitions), supporting cycles, human-in-the-loop, and persistence.

langgraph_agent.py

Python

1	from typing import TypedDict, Annotated
2	from langgraph.graph import StateGraph, END
3	from langgraph.graph.message import add_messages
4	from langchain_openai import ChatOpenAI
5	from langchain_core.messages import HumanMessage
6
7	class AgentState(TypedDict):
8	messages: Annotated[list, add_messages]
9
10	llm = ChatOpenAI(model="gpt-4o", temperature=0)
11
12	def chatbot(state: AgentState):
13	return {"messages": [llm.invoke(state["messages"])]}
14
15	def should_continue(state: AgentState):
16	last_message = state["messages"][-1]
17	if hasattr(last_message, "tool_calls") and last_message.tool_calls:
18	return "tools"
19	return END
20
21	# Build the graph
22	graph = StateGraph(AgentState)
23	graph.add_node("chatbot", chatbot)
24	graph.set_entry_point("chatbot")
25	graph.add_conditional_edges("chatbot", should_continue)
26	graph.compile()
27
28	# Run the agent
29	from langgraph.checkpoint.memory import MemorySaver
30	memory = MemorySaver()
31	app = graph.compile(checkpointer=memory)
32
33	config = {"configurable": {"thread_id": "1"}}
34	result = app.invoke(
35	{"messages": [HumanMessage(content="What is 2+2?")]},
36	config=config,
37	)
38	print(result["messages"][-1].content)

✓

best practice

Use LangGraph for any agent that needs cycles, conditional branching, or state persistence. The create_react_agentshortcut works for simple cases, but LangGraph gives you full control over the agent's decision-making flow.

Callbacks & Tracing

Callbacks hook into every step of a chain or agent execution — prompt formatting, LLM calls, tool invocations — giving you observability without modifying your pipeline. LangSmith provides hosted tracing, dashboards, and evaluation.

custom_callback.py

Python

1	# Custom callback handler — log every LLM call
2	from langchain_core.callbacks import BaseCallbackHandler
3	from langchain_core.outputs import LLMResult
4	import time
5
6	class TimingCallback(BaseCallbackHandler):
7	def __init__(self):
8	self.calls = []
9
10	def on_llm_start(self, serialized, prompts, **kwargs):
11	self._start_time = time.time()
12	print(f"LLM call started — {len(prompts)} prompt(s)")
13
14	def on_llm_end(self, response: LLMResult, **kwargs):
15	elapsed = time.time() - self._start_time
16	tokens = response.llm_output.get("token_usage", {})
17	self.calls.append({
18	"elapsed": round(elapsed, 2),
19	"input_tokens": tokens.get("prompt_tokens", 0),
20	"output_tokens": tokens.get("completion_tokens", 0),
21	})
22	print(f"LLM call done in {elapsed:.2f}s — {tokens}")
23
24	handler = TimingCallback()
25	result = chain.invoke(
26	{"question": "Explain recursion"},
27	config={"callbacks": [handler]},
28	)
29	print(f"Total calls: {len(handler.calls)}")

langsmith_tracing.py

Python

1	# LangSmith tracing — one env var enables full tracing
2	import os
3	os.environ["LANGSMITH_TRACING"] = "true"
4	os.environ["LANGSMITH_API_KEY"] = "ls_..."
5
6	# All chains, agents, and tools are now traced automatically
7	# View traces at smith.langchain.com
8
9	# Add tags and metadata for filtering in the dashboard
10	from langchain_core.runnables import RunnableConfig
11
12	config = RunnableConfig(
13	tags=["production", "customer-support"],
14	metadata={"user_id": "u_12345", "session_id": "s_67890"},
15	run_name="Customer Query Handler",
16	)
17
18	result = chain.invoke(
19	{"question": "Where is my order?"},
20	config=config,
21	)
22
23	# Custom project per chain for organized dashboards
24	os.environ["LANGCHAIN_PROJECT"] = "forgelearn-prod"

✓

best practice

Enable LangSmith tracing during development — it's the fastest way to debug chain failures and understand token usage patterns. Use RunnableConfig tags to filter traces by environment, feature, or user segment.

Production Patterns

Building production-ready LangChain applications requires attention to error handling, observability, cost control, and deployment patterns.

production.py

Python

1	# Production-ready chain with error handling and observability
2	from langchain_openai import ChatOpenAI
3	from langchain_core.prompts import ChatPromptTemplate
4	from langchain_core.output_parsers import StrOutputParser
5	from langchain_core.runnables import RunnableConfig
6	import logging
7
8	logger = logging.getLogger(__name__)
9
10	llm = ChatOpenAI(model="gpt-4o", temperature=0, max_retries=2, request_timeout=30)
11
12	# Add retries and callbacks
13	config = RunnableConfig(
14	callbacks=[logging_handler], # Your custom callback handler
15	tags=["production", "v2"],
16	metadata={"environment": "production"},
17	)
18
19	prompt = ChatPromptTemplate.from_template(
20	"Answer concisely: {question}"
21	)
22	chain = prompt \| llm \| StrOutputParser()
23
24	# Try/except with fallback
25	try:
26	result = chain.invoke(
27	{"question": "What is Docker?"},
28	config=config,
29	)
30	except Exception as e:
31	logger.error(f"Chain failed: {e}")
32	result = "I apologize, but I'm unable to answer right now."
33
34	# Streaming for long responses
35	async def stream_answer(question: str):
36	full_response = ""
37	async for chunk in chain.astream(
38	{"question": question},
39	config=config,
40	):
41	full_response += chunk
42	yield chunk
43	logger.info(f"Response length: {len(full_response)} chars")

cost_control.py

Python

1	# Cost control with token counting
2	from langchain_core.callbacks import get_usage_metadata
3
4	def get_token_usage(chain, input_data):
5	"""Invoke chain and return token usage metrics."""
6	result = chain.invoke(input_data)
7	usage = get_usage_metadata()
8	return {
9	"result": result,
10	"input_tokens": usage.get("input_tokens", 0),
11	"output_tokens": usage.get("output_tokens", 0),
12	"total_cost_usd": (
13	usage.get("input_tokens", 0) * 0.0025 / 1000
14	+ usage.get("output_tokens", 0) * 0.01 / 1000
15	),
16	}

⚠

warning

Never use temperature=0 for agent tool-calling — it can cause the model to skip tool calls. Use temperature=0 only for pure text generation tasks where determinism is critical.

$Blueprint — Engineering Documentation·Section ID: AI-LC-01·Revision: 1.0

X GitHub YouTube Instagram Facebook Discord Telegram Buy me a coffee Sponsor