10 Essential LangChain & LangGraph Concepts Every AI Engineer Must Master

Why Architecture Matters

Most AI demos stop at a simple question‑answer loop, but production agents must retrieve documents, call tools, handle failures, route requests, preserve history, and incorporate human review. The gap between demo and production is often due to architecture, not the model itself. LangChain and LangGraph are the most widely adopted frameworks that provide the building blocks for robust AI systems.

1. State – Shared Memory for Agents

Production agents need a shared state rather than isolated prompts. A support‑ticket example defines a TypedDict with fields such as ticket_id, customer_message, retrieved_articles, and sentiment_score. The workflow updates this dictionary at each step, giving the system predictable context.

from typing import TypedDict

class SupportState(TypedDict):
    ticket_id: str
    customer_message: str
    customer_tier: str
    retrieved_articles: list
    sentiment_score: float
    draft_response: str
    escalation_required: bool
    final_response: str

{
    "ticket_id": "TKT-9012",
    "customer_message": "My payment failed twice",
    "customer_tier": "Enterprise",
    "retrieved_articles": ["Payment Retry Policy", "Credit Card Validation"],
    "sentiment_score": 0.91,
    "draft_response": "Please retry the payment...",
    "escalation_required": false
}

2. Node – Just a Function

Nodes are ordinary Python functions. Examples include a retrieval node, a re‑ranking node, and a generation node.

def retrieve_documents(state):
    docs = vector_store.similarity_search(
        state["user_query"],
        k=5
    )
    return {"retrieved_docs": docs}

def rerank_documents(state):
    reranked = reranker.rank(
        query=state["user_query"],
        docs=state["retrieved_docs"]
    )
    return {"retrieved_docs": reranked[:3]}

def generate_answer(state):
    response = llm.invoke(state["retrieved_docs"])
    return {"draft_answer": response}

These nodes are then composed in a linear sequence:

Retrieval
    ↓
Re‑ranking
    ↓
Generation
    ↓
Verification

3. Chain vs Graph

Chains work for simple linear flows, but graphs handle multiple agents, conditional branches, human approval, retries, and long‑running tasks. An ERP migration assistant illustrates a graph with separate research and support agents converging to a final response.

4. Routing Over Long Prompts

Instead of a monolithic prompt that tries to be an expert in everything, classify the user query and route it to a specialized agent.

def classify_request(state):
    query = state["user_query"]
    if "billing" in query:
        return "billing_agent"
    if "error" in query:
        return "technical_agent"
    if "pricing" in query:
        return "sales_agent"
    return "general_agent"

The routing diagram shows the user request flowing into three parallel agents (technical, billing, sales) before producing a final answer.

5. Retrieval – More Than Vector Search

Production RAG pipelines add re‑ranking, filtering, and context compression after the initial vector search.

docs = vector_store.search(query, k=20)
reranked_docs = reranker.rank(query, docs)
top_docs = reranked_docs[:5]

In a 10‑million‑document knowledge base, only a handful of the retrieved documents contain the needed information, making re‑ranking critical for answer quality.

6. Structured Outputs

Returning raw JSON often leads to parsing errors. Defining a Pydantic model forces the LLM to emit validated data.

from pydantic import BaseModel

class TicketClassification(BaseModel):
    category: str
    severity: str
    assigned_team: str
    confidence_score: float

result = llm.with_structured_output(TicketClassification).invoke(ticket_text)

{
    "category": "Payment",
    "severity": "High",
    "assigned_team": "Billing",
    "confidence_score": 0.94
}

7. Streaming Output

Streaming tokens improves perceived performance. A simple loop prints each chunk, and more complex workflows can yield progress messages.

for chunk in model.stream(prompt):
    print(chunk)

yield "Analyzing requirements…"
yield "Retrieving documents…"
yield "Generating response…"
yield "Validating result…"
yield final_response

8. Memory Beyond Chat History

Operational memory stores repository state, modified files, failed tests, etc., enabling the agent to make informed decisions across multiple steps.

{
    "repository": "payments-api",
    "modified_files": ["payment.service.ts", "retry.handler.ts"],
    "failed_tests": ["payment_retry.spec.ts"],
    "last_fix_attempt": "...",
    "pull_request_url": "..."
}

9. Checkpointing for Long‑Running Workflows

Checkpoints allow a workflow to resume after failures such as model timeouts or unavailable APIs.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

During a loan‑approval process, checkpoints are placed after document collection, risk assessment, and before human approval, ensuring progress is not lost.

10. Human‑in‑the‑Loop

High‑risk decisions still require human oversight. An approval node returns either "human_review" or "auto_approve" based on a risk score.

def approval_node(state):
    if state["risk_score"] > 0.8:
        return "human_review"
    return "auto_approve"

The workflow proceeds through AI analysis, risk evaluation, optional human review, and then continues.

Conclusion

Repeatedly, the article finds that reliable AI applications depend far more on workflow design than on model choice. Mastering the ten concepts above equips AI engineers to build systems that scale, remain maintainable, and integrate smoothly with human processes.

State

Node

Graph

Routing

Retrieval

Structured Outputs

Memory

Checkpointing

Human Review