Getting Started with LangChain: Building Production-Ready LLM & Agent Pipelines

The rapid evolution of Large Language Models (LLMs) has shifted software development from static rule-based systems to dynamic, context-aware AI applications. However, raw API calls to LLMs are rarely sufficient for production. Real-world applications demand structured input validation, stateful memory, retrieval from proprietary vector databases, tool invocation, and multi-step autonomous decision loops.

LangChain is the industry-standard orchestration framework that unifies these primitives into composable, observable pipelines using the LangChain Expression Language (LCEL) and LangGraph.

In this guide, we will step through constructing clean LCEL chains, integrating RAG (Retrieval-Augmented Generation), binding custom tools, and building autonomous agent workflows.

1. Environment & Core Dependencies

Let’s set up the core ecosystem packages:

# Install core LangChain packages and community model providers
pip install langchain langchain-core langchain-community langchain-openai chromadb

Export your API keys into your environment:

export OPENAI_API_KEY="your-api-key-here"

2. The Foundation: LCEL (LangChain Expression Language)

Modern LangChain relies on LCEL, which treats components as declarative pipelines using the unix pipe operator (|). Every component in an LCEL chain implements the Runnable protocol, automatically enabling synchronous execution (invoke), streaming (stream), asynchronous support (ainvoke), and batch processing (batch).

Here is a minimal, structured translation pipeline:

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

# 1. Define Model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

# 2. Define Prompt Template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert technical translator. Translate the given text into {target_language} with precise engineering terminology."),
    ("user", "{text}")
])

# 3. Define Parser
parser = StrOutputParser()

# 4. Compose the Chain via LCEL
chain = prompt | model | parser

# 5. Execute
result = chain.invoke({
    "target_language": "German",
    "text": "The distributed database ensures eventual consistency across multiple edge nodes."
})

print("Translation:", result)

3. Structured Outputs with Pydantic

Production systems cannot rely on unpredictable raw string responses. LangChain allows you to bind Pydantic schemas directly to models to guarantee deterministic JSON output:

from pydantic import BaseModel, Field
from typing import List

class CodeReviewSummary(BaseModel):
    summary: str = Field(description="High-level overview of the code quality")
    severity: str = Field(description="Risk rating: LOW, MEDIUM, HIGH, or CRITICAL")
    suggested_fixes: List[str] = Field(description="Concrete refactoring recommendations")
    estimated_refactor_minutes: int = Field(description="Time required to apply fixes")

# Bind schema to the model
structured_llm = model.with_structured_output(CodeReviewSummary)

review_prompt = ChatPromptTemplate.from_messages([
    ("system", "Analyze the provided Python snippet for concurrency bugs, memory leaks, and style issues."),
    ("user", "{code}")
])

review_chain = review_prompt | structured_llm

report = review_chain.invoke({
    "code": "def process_data(items=[]): items.append(1); return items"
})

print("Severity:", report.severity)
print("Fixes:", report.suggested_fixes)

4. Retrieval-Augmented Generation (RAG)

When models lack private context or real-time internal data, RAG retrieves relevant document chunks from a vector database and injects them into the prompt before generation.

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

# 1. Seed vector store with documents
docs = [
    Document(page_content="LPU policy dictates that student hackathon leaves must be sanctioned 48 hours in advance by the HOD."),
    Document(page_content="Exam re-evaluations must be submitted through the university portal within 7 working days of result declaration."),
]

vectorstore = Chroma.from_documents(docs, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})

# 2. Define RAG Prompt
rag_prompt = ChatPromptTemplate.from_template("""
Answer the question based strictly on the following provided context:
<context>
{context}
</context>

Question: {question}
""")

# Helper to format retrieved documents
def format_docs(documents):
    return "\n\n".join(doc.page_content for doc in documents)

# 3. Compose RAG Chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("When must a hackathon leave be submitted?")
print("RAG Answer:", answer)

5. Building Autonomous Tool-Calling Agents

The pinnacle of modern AI engineering is building Agentic systems where the LLM decides which tool to call, executes it, observes the result, and loops until the goal is satisfied.

from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 1. Define custom tools
@tool
def calculate_student_gpa(grades: list[float]) -> float:
    """Calculates the grade point average given a list of numeric course grades."""
    return round(sum(grades) / len(grades), 2)

@tool
def lookup_academic_calendar(event: str) -> str:
    """Returns official dates for academic milestones like exams and holidays."""
    calendar = {
        "midterm": "October 14 - October 22, 2026",
        "endterm": "December 01 - December 18, 2026",
    }
    return calendar.get(event.lower(), "Event not found in academic calendar.")

tools = [calculate_student_gpa, lookup_academic_calendar]

# 2. Agent Prompt with intermediate step placeholders
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an intelligent university AI assistant with access to official calculation and lookup tools."),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# 3. Create Agent and Executor
agent = create_tool_calling_agent(model, tools, agent_prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 4. Invoke multi-step agent query
response = executor.invoke({
    "input": "Can you check when midterms are, and also compute my GPA for grades [8.5, 9.0, 7.8, 8.2]?"
})

print("Agent Response:\n", response["output"])

Key Takeaways for Production AI

1. Deterministic Contracts: Always use with_structured_output with Pydantic for API layers.
2. State Management: For complex multi-turn workflows, branch out from linear chains to graph-based state machines using LangGraph.
3. Observability: Integrate LangSmith or OpenTelemetry early to trace token latency, tool invocation failures, and prompt drift.