Advanced Data Structures & Algorithms

Java Backend Modernization

Beginner's Guide to Data Structures

Agentic AI Systems

Data Engineering & Spark Mastery

Low-Level Design Interview Prep

Microservices Interview Roadmap

Design Patterns Interview Roadmap

Distributed Systems Interview Roadmap

Mastering Python Development

System Design Interview Prep

Distributed Systems Refresher

AI Engineering Launchpad

Architecture Patterns for Scalable Systems

RAG & Vector Search Mastery

Production DevOps & Reliability

Enterprise Patterns Playbook

Master Coding Interviews

Coding Techniques Interview Roadmap

Data Structures Interview Roadmap

Architecture Patterns Interview Roadmap

software-engineering-principles

Migrate your Spring Boot services from thread-pool exhaustion to JDK 21 virtual threads — the practical Spring developer guide

Adapting to Virtual Threads for Spring Developers

how-it-works-internals-explained

From WAL and MVCC to fsync and isolation levels — the internals behind every ACID guarantee in PostgreSQL, MySQL, Oracle, and SQL Server.

ACID Properties Explained: How SQL Databases Guarantee Atomicity, Consistency, Isolation, and Durability

How modern NoSQL databases achieve atomicity and isolation without a single machine

ACID Transactions in Distributed Databases: DynamoDB, Cosmos DB, and Spanner Compared

llm-engineering

Is an LLM a brain in a jar? To make it truly useful, we need to give it access to the world. This guide explains RAG and Agents.

Advanced AI: Agents, RAG, and the Future of Intelligence

Production AI needs explicit routing, memory, execution, and evaluation layers rather than one loop.

AI Architecture Patterns: Routers, Planner-Worker Loops, Memory Layers, and Evaluation Guardrails

datastructures-and-algorithms

Master shortest paths, dependency resolution, and cycle detection with robust Java solutions.

Advanced Graph Algorithms: Dijkstra, Topological Sort, and Disjoint Set Union (DSU)

An LLM can talk, but an AI Agent can *act*. We explain how Agents use the ReAct framework to brow...

AI Agents Explained: When LLMs Start Using Tools

A production-first guide to picking the right vector index for recall, latency, memory, and update patterns.

ANN Index Types Explained: When to Choose Flat, HNSW, IVF, or IVF-PQ

apache-spark-engineering

Process petabytes with Python: understand Spark's execution model, DataFrame transformations, partitioning strategy, and real-time streaming.

Apache Spark for Data Engineers: RDDs, DataFrames, and Structured Streaming

ACID guarantees consistency. BASE guarantees availability. We explain the trade-offs between SQL...

BASE Theorem Explained: How it Stands Against ACID

machine-learning-fundamentals

From the query-key-value formulation to multi-head attention — the mechanism that made transformers dominate AI

Attention Mechanism Explained: How Transformers Learn to Focus

system-design-interview-prep

They all sit in front of your servers. But do you need Nginx, HAProxy, or Kong? We clarify the roles of each component.

API Gateway vs. Load Balancer vs. Reverse Proxy: What's the Difference?

architecture-patterns-for-production-systems

Mobile apps need less data than Desktop dashboards. Why serve them the same JSON? The Backend for Frontend (BFF) pattern solves this.

Backend for Frontend (BFF): Tailoring APIs for UI

BFS explores graphs level by level using a queue — the key to shortest paths and level-order problems.

BFS — Breadth-First Search: Level-by-Level Graph Exploration

Backtracking is a brute-force algorithmic technique that solves problems by trying to build a sol...

Exploring Backtracking Techniques in Data Structures

big-data-engineering

Choose ingestion, serving, and ownership patterns deliberately when data platforms start to scale.

Big Data Architecture Patterns: Lambda, Kappa, CDC, Medallion, and Data Mesh

Why traditional databases fail at scale, what the 5 Vs really mean, and a practical map of the big data ecosystem from ingestion to insights.

Big Data 101: The 5 Vs, Ecosystem, and Why Scale Breaks Everything

Not all Binary Trees are created equal. We explain Full, Complete, Perfect, and Balanced Binary T...

Exploring Different Types of Binary Trees

Master Big O notation: measure algorithm efficiency from O(1) to O(n²) with growth diagrams, worked examples, and the interview mental model.

Big O Notation Explained: Time Complexity, Space Complexity, and Why They Matter in Interviews

Binary search has five distinct patterns — classic, leftmost, rightmost, rotated arrays, and 2D matrices.

Binary Search Patterns: Five Variants Every Senior Engineer Knows

Simple AI agents react one step at a time. Multistep agents are different: they create a full pla...

Multistep AI Agents: The Power of Planning

Stop trying to solve everything with ChatGPT. We provide a decision framework for modern develope...

'The Developer''s Guide: When to Use Code, ML, LLMs, or Agents'

A practical field guide to prompt spaghetti, missing evals, runaway costs, and the moment an LLM feature becomes too much.

LLM Software Development Pitfalls: What to Avoid and When to Simplify

Is Prompt Engineering a real skill? Yes. We explore the science behind talking to AI. Learn how Z...

Prompt Engineering Guide: From Zero-Shot to Chain-of-Thought

agentic-ai-langchain-and-langgraph

Build your first LangChain pipeline: LCEL, ChatPromptTemplate, and LLM-agnostic model swapping from OpenAI to Anthropic to Ollama.

LangChain 101: Chains, Prompts, and LLM Integration

Keep context across turns: ConversationBufferMemory, ConversationSummaryMemory, and the LCEL memory pattern — before LangGraph checkpointing.

LangChain Memory: Conversation History and Summarization

Build tool-using agents with LangChain: @tool decorator, AgentExecutor, ReAct reasoning, and when to reach for LangGraph instead.

LangChain Tools and Agents: The Classic Agent Loop

LangChain's AgentExecutor handles simple loops — but stateful branching, long-running tasks, and human-in-the-loop require LangGraph's graph model.

From LangChain to LangGraph: When Agents Need State Machines

Learn LangGraph from scratch: StateGraph, typed state, nodes, edges, and your first LLM-agnostic stateful agent.

LangGraph 101: Building Your First Stateful Agent

Build ReAct agents in LangGraph: prebuilt create_react_agent vs custom StateGraph loop, with a coding assistant example.

The ReAct Agent Pattern in LangGraph: Think, Act, Observe, Repeat

Wire real capabilities into LangGraph agents: @tool decorator, ToolNode, bind_tools, parallel execution, and error handling.

LangGraph Tool Calling: ToolNode, Parallel Tools, and Custom Tools

Pause LangGraph agents mid-run for human approval: interrupt(), Command, update_state(), and async resume patterns.

Human-in-the-Loop Workflows with LangGraph: Interrupts, Approvals, and Async Execution

Give LangGraph agents persistent memory: checkpointers, thread IDs, cross-session memory store, and context overflow strategies.

LangGraph Memory and State Persistence: Checkpointers, Threads, and Cross-Session Memory

Build multi-agent systems in LangGraph: supervisor routing, worker handoffs, subgraphs, and the Send API for parallel agents.

Multi-Agent Systems in LangGraph: Supervisor Pattern, Handoffs, and Agent Networks

Deploy LangGraph agents to production: LangServe, Docker, PostgresSaver, LangGraph Platform, and LangSmith observability.

Deploying LangGraph Agents: LangServe, Docker, LangGraph Platform, and Production Observability

Tools do one action; skills orchestrate many steps. Learn why this distinction makes agents far more reliable.

LLM Skills vs Tools: The Missing Layer in Agent Design

After tools and skills, this is the control plane: registry design, routing rules, and evaluation loops.

LLM Skill Registries, Routing Policies, and Evaluation for Production Agents

## Scenario context

A standard LLM is a brain in a jar — it can reason but cannot act. An **AI Agent** connects that brain to tools (web search, code execution, APIs). Instead of just answering a question, an agent executes a loop of Thought → Action → Observation until the goal is reached.

---

An **AI agent** is a program that uses a large language model as its reasoning engine and connects it to **external tools** so it can take re…

This walkthrough focuses on **The Basics: What Is an AI Agent**.

Before making a design choice, clarify:

- An AI agent is a program that uses a large language model as its reasoning engine and connects it to external tools so it can take real actions in the world.
- Where a plain LLM is purely generative — text in, text out — an agent augments the LLM with capabilities like web search, code execution, file reading, database queries, and API calls.
- Three components make up almost every agent: 1.

**Concepts in play:** llm-agents-and-langgraph, sorting-and-searching-algorithms, llm-prompt-engineering, database-indexing

## Stage 1 — The ReAct Loop: Thought → Action → Observation

- The dominant pattern for agents is ReAct (Reasoning + Acting).
- The model cycles through three steps until the task is complete: This loop continues until the model decides it has enough information to answer.

*Applies **llm-agents-and-langgraph** to the decision above.*

```mermaid
flowchart TD
    Start([User Goal]) --> T[Thought: What do I need?]
    T --> A[Action: Call a Tool]
    A --> O[Observation: Tool Result]
    O --> D{Goal reached?}
    D -- No --> T
    D -- Yes --> Answer([Return Final Answer])
```

```mermaid
sequenceDiagram
    participant U as User
    participant L as LLM
    participant T as Tool
    U->>L: User request
    L-->>L: Thought: what to do
    L->>T: Action: call tool
    T-->>L: Observation: result
    L-->>L: Next thought
    L-->>U: Final Answer
```

**Observable outcome:** The model cycles through three steps until the task is complete: This loop continues until the model decides it has enough information to answer.

## Stage 2 — How an Agent Processes a Request

- When a user sends a question to an agent-powered system, several layers of logic activate before any answer is returned.
- Understanding this flow helps you reason about latency, costs, and failure points.
- What happens at each node: - Inject Tool Schemas — before the first LLM call, all tool names, descriptions, and input types are serialised into the system prompt so the model knows what it can use.
- - LLM Reasoning — the model outputs a structured "Thought" explaining its next move.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```mermaid
flowchart TD
    U([User Request]) --> SYS[Inject Tool Schemas into LLM Context]
    SYS --> LLM1[LLM Reasoning: Thought  What do I need?]
    LLM1 --> TC[Tool Selection: Which tool and args?]
    TC --> TE[Tool Execution: Call external function]
    TE --> OBS[Observation: Capture tool output]
    OBS --> CHK{Goal reached?}
    CHK -- No --> LLM1
    CHK -- Yes --> ANS([Return Final Answer to User])
```

**Observable outcome:** - LLM Reasoning — the model outputs a structured "Thought" explaining its next move.

## Stage 3 — Tool Definitions: How an Agent Knows What It Can Do

- A tool is a function the model can call.
- In LangChain you define tools with a name, description, and input schema: The model receives the tool descriptions in its system prompt and decides which to call (and with what arguments) based on the task.
- It never sees the implementation — only the name and docstring.

*Applies **llm-prompt-engineering** to the decision above.*

```python
from langchain.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for current information. Use this for recent events or facts."""
    return web_search_api(query)

@tool
def run_python(code: str) -> str:
    """Execute Python code and return the output. Use this for calculations."""
    return exec_sandbox(code)
```

```mermaid
flowchart TD
    I[Input from User] --> LM[LLM Reasoning]
    LM --> D{Need a tool?}
    D -- No --> R[Return Answer]
    D -- Yes --> TS[Select Tool]
    TS --> EX[Execute Tool]
    EX --> OB[Observe Output]
    OB --> LM
```

**Observable outcome:** It never sees the implementation — only the name and docstring.

## Stage 4 — Deep Dive: Building a Simple Agent with LangChain

- The verbose=True flag shows you the full Thought/Action/Observation chain — invaluable for debugging.

*Applies **database-indexing** to the decision above.*

```python



from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain import hub

llm = ChatOpenAI(model="gpt-4o-mini")
tools = [search_web, run_python]

prompt = hub.pull("hwchase17/react")          # standard ReAct prompt
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({"input": "Who was US president when Titanic was released?"})
print(result["output"])
```

**Observable outcome:** The verbose=True flag shows you the full Thought/Action/Observation chain — invaluable for debugging.

## Stage 5 — Real-World Applications: Real-World Agent Use Cases

- 

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** You can explain how “Real-World Applications: Real-World Agent Use Cases” changes the architecture or implementation path.

## Stage 6 — Practical: Debugging Agent Behavior

- Agents fail in opaque ways that are hard to catch without systematic debugging.
- Here is a practical workflow for diagnosing misbehaving agents.
- Step 1 — Enable verbose output.
- In LangChain, set verbose=True on the AgentExecutor .

*Applies **sorting-and-searching-algorithms** to the decision above.*

```python
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)
```

**Observable outcome:** In LangChain, set verbose=True on the AgentExecutor .

## End-to-end outcome

An AI agent is a program that uses a large language model as its reasoning engine and connects it to external tools so it can take real actions in the world.

## Alternatives considered

### Brain in a Jar vs Brain with Arms

Give it "What is the weather in Tokyo today?" and it will: - Answer from training data (which is months or years old).

### Decision Guide: When to Use an AI Agent

Use an agent when the task requires multiple steps, tool use, or dynamic decision-making that can't be scripted upfront.


## Failure and degraded modes

### Trade-offs & Failure Modes: When Agents Fail: Hallucinations, Loops, and Cost Blowouts

- **Detection:** Agents introduce new failure modes beyond plain LLMs: Hallucinated tool calls — the model invents arguments or calls a non-existent tool.
- **User or operator impact:** Agents introduce new failure modes beyond plain LLMs: Hallucinated tool calls — the model invents arguments or calls a non-existent tool.



## Scenario context

A simple ReAct agent reacts one tool call at a time. A **multistep agent** plans a complete task decomposition upfront, then executes each step sequentially — handling complex goals that require 5-10 interdependent actions without re-prompting the LLM for each step.

---

A **line cook** (simple ReAct agent): receives one ticket, cooks one dish, hands it over. Then the next ticket.

A **head chef** (multistep agent)…

This walkthrough focuses on **Line Cook vs. Head Chef**.

Before making a design choice, clarify:

- A line cook (simple ReAct agent): receives one ticket, cooks one dish, hands it over.
- A head chef (multistep agent): receives the full dinner party menu, plans the entire 5-course sequence, coordinates prep timing for all dishes, anticipates which items can be done in parallel, and manages the full execu…
- The difference: planning before acting .

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, stack-and-queue-data-structures, api-security-oauth2

## Stage 1 — What Makes a Multistep Agent Different

- A simple ReAct agent operates with a single-step planning horizon : it looks at the current state, picks the next action, observes the result, and repeats.
- This works well for short, open-ended tasks where the required steps are unknown ahead of time.
- But when a task involves 5–10 interdependent actions, this approach becomes inefficient — the LLM must rediscover context at every step, decisions made in step 6 can conflict with assumptions from step 2, and there is n…
- A multistep agent solves this by introducing a distinct planning phase before any tool is called.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** A multistep agent solves this by introducing a distinct planning phase before any tool is called.

## Stage 2 — The Plan-and-Execute Architecture

- Goal: "Research the top 3 AI papers from last month, summarize each, and draft a blog post." This flow highlights two key design decisions: plan validation happens before any tool call (catching hallucinated tool names…
- The re-plan path preserves completed step results — there is no need to restart from the beginning when a single step fails mid-execution.

*Applies **graph-traversal-algorithms** to the decision above.*

```mermaid
flowchart TD
    Goal[User Goal] --> Planner["LLM Planner (produce JSON step list)"]
    Planner --> Validate["Validate Plan (tool names, schema)"]
    Validate -->|Valid| Decompose[Decompose into Parallel + Sequential Steps]
    Validate -->|Invalid| Planner
    Decompose --> Parallel["Run Parallel Steps (independent actions)"]
    Decompose --> Sequential["Run Sequential Steps (dependent actions)"]
    Parallel --> Aggregate[Aggregate Intermediate Results]
    Sequential --> Aggregate
    Aggregate --> Check{All steps complete?}
    Check -->|Yes| Final[Final Result Delivered]
    Check -->|Step Failed| Replan[Re-plan Remaining Steps from Current State]
    Replan --> Decompose
```

**Observable outcome:** The re-plan path preserves completed step results — there is no need to restart from the beginning when a single step fails mid-execution.

## Stage 3 — Real-World Applications: Real-World Use Cases

- Multistep agents are not theoretical constructs — they power several categories of real production systems today.
- Automated Research Pipelines A research agent might receive a goal like "produce a competitive analysis of the top 5 CRM tools." The plan includes: search for the top 5 tools (step 1), fetch the pricing page for each to…
- No human re-prompts the system between steps — the plan drives the full execution.
- Coding Agents Agents like GitHub Copilot Workspace and Devin use multistep planning to handle tasks like "add OAuth login to this codebase." The plan might include: reading existing auth files, identifying integration p…

*Applies **stack-and-queue-data-structures** to the decision above.*

```mermaid
sequenceDiagram
    participant U as User
    participant P as LLM Planner
    participant E as Executor
    participant T as Tools

    U->>P: Submit goal
    P->>P: Decompose into step list
    P-->>E: Plan [step1, step2, step3...]
    E->>T: Execute step 1
    T-->>E: Result 1
    E->>P: Observe result 1
    P->>P: Update plan if needed
    E->>T: Execute step 2
    T-->>E: Result 2
    E->>T: Execute step 3
    T-->>E: Result 3
    E-->>U: Aggregated final answer
```

```mermaid
flowchart TD
    Goal[Receive Goal]
    Decompose[LLM: Decompose into step list]
    Queue["Queue Steps (DAG order)"]
    Execute[Execute Next Step with Tools]
    UpdateState[Update Shared State with Result]
    Done{All Steps Complete?}
    Replan{Step Failed?}
    Final[Return Final Result]
    ReplanStep[Re-plan Remaining Steps from State]

    Goal --> Decompose --> Queue --> Execute --> UpdateState
    UpdateState --> Done
    Done -->|Yes| Final
    Done -->|No| Replan
    Replan -->|Yes| ReplanStep --> Queue
    Replan -->|No| Execute
```

**Observable outcome:** Coding Agents Agents like GitHub Copilot Workspace and Devin use multistep planning to handle tasks like "add OAuth login to this codebase." The plan might include: reading existing auth files, identifying integration p…

## Stage 4 — Practical: Building a Simple Plan-and-Execute Agent

- The fastest way to understand multistep agents is to trace how the planner and executor components interact in code.
- Below is a minimal Python example using LangChain's PlanAndExecute abstraction with custom tools.
- Step 1 — Define tools and the LLM: Step 2 — Wire up planner and executor: Step 3 — Invoke with a multi-step goal: When you run this with verbose=True , you will see the planner emit a JSON step list, then the executor c…
- This trace is invaluable for debugging — if a step produces an unexpected output, you can inspect exactly where the plan diverged from reality and adjust either the planner prompt or the tool implementation accordingly.

*Applies **api-security-oauth2** to the decision above.*

```python
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for the given query."""
    return f"Results for: {query}"

@tool
def summarize_text(text: str) -> str:
    """Summarize the provided text."""
    return f"Summary of: {text[:50]}..."

tools = [search_web, summarize_text]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
```

```python
from langchain_experimental.plan_and_execute import (
    PlanAndExecute, load_agent_executor, load_chat_planner
)

planner  = load_chat_planner(llm)                          # single LLM call that generates the JSON step list
executor = load_agent_executor(llm, tools, verbose=True)   # runs each step; verbose=True prints tool call traces
agent    = PlanAndExecute(planner=planner, executor=executor, verbose=True)
```

**Observable outcome:** This trace is invaluable for debugging — if a step produces an unexpected output, you can inspect exactly where the plan diverged from reality and adjust either the planner prompt or the tool implementation accordingly.

## Stage 5 — LangGraph and AutoGen: How OSS Frameworks Implement Multistep Agent Orchestration

- Two open-source frameworks have emerged as the leading ways to build multistep agents with explicit state management, and they take very different architectural approaches.
- LangGraph is a graph-based agent orchestration library from LangChain that models an agent's execution as a directed graph of nodes (LLM calls, tool calls, or logic) and edges (conditional transitions).
- It gives you full control over state, supports cycles (loops), and is designed for durable, resumable workflows — making it the production-ready choice for complex multistep pipelines.
- AutoGen (Microsoft) takes a conversation-centric approach — agents are modeled as participants in a multi-agent dialogue, where each agent has a role (AssistantAgent, UserProxyAgent) and can call tools or other agents a…

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langgraph.graph import StateGraph, END
from typing import TypedDict

# Define the shared state for the agent
class AgentState(TypedDict):
    task: str
    plan: list
    results: list
    current_step: int

# Build the graph
graph = StateGraph(AgentState)

# Add nodes: each is a function that reads and writes state
graph.add_node("planner", run_planner)      # LLM call → produces plan list
graph.add_node("executor", run_executor)    # Tool call → appends to results
graph.add_node("checker", check_completion) # Logic → routes to next step or END

# Wire the execution flow
graph.set_entry_point("planner")
graph.add_edge("planner", "executor")
graph.add_conditional_edges(
    "executor",
    lambda state: "checker" if state["current_step"] < len(state["plan"]) else END,
)
graph.add_edge("checker", "executor")

agent = graph.compile()
result = agent.invoke({"task": "Research top 3 AI papers and draft a summary", "plan": [], "results": [], "current_step": 0})
```

```python
from autogen import AssistantAgent, UserProxyAgent

planner = AssistantAgent(
    name="Planner",
    system_message="You decompose goals into numbered steps and produce a JSON plan.",
    llm_config={"model": "gpt-4o"},
)

executor = UserProxyAgent(
    name="Executor",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "./workspace"},
)

# The executor drives the conversation; the planner produces the plan
executor.initiate_chat(
    planner,
    message="Research the top 3 AI papers from last month and summarize each.",
    max_turns=10,
)
```

**Observable outcome:** AutoGen (Microsoft) takes a conversation-centric approach — agents are modeled as participants in a multi-agent dialogue, where each agent has a role (AssistantAgent, UserProxyAgent) and can call tools or other agents a…

## Stage 6 — Key Lessons

- Plan validation is non-negotiable.
- Always validate the generated plan against the registered tool list and expected argument schema before execution begins.
- Catching hallucinated tool names at plan time costs one validation pass; catching them at execution time costs the tokens from all preceding steps plus a re-plan.
- Re-plan from failure state, not from scratch.

*Applies **graph-traversal-algorithms** to the decision above.*

**Observable outcome:** Re-plan from failure state, not from scratch.

## End-to-end outcome

A line cook (simple ReAct agent): receives one ticket, cooks one dish, hands it over.

## Alternatives considered

### Line Cook vs. Head Chef

A line cook (simple ReAct agent): receives one ticket, cooks one dish, hands it over.

### Simple ReAct vs. Plan-and-Execute: Core Difference

Simple ReAct vs. Plan-and-Execute: Core Difference

### Decision Guide: ReAct vs. Plan-and-Execute

Decision Guide: ReAct vs. Plan-and-Execute


## Failure and degraded modes

### Trade-offs & Failure Modes: When to Use Multistep Agents vs Simple Agents

- **Detection:** Critical failure modes for multistep agents: - Stale plan: If step 3 fails, steps 4-8 may be based on incorrect assumptions.
- **User or operator impact:** Critical failure modes for multistep agents: - Stale plan: If step 3 fails, steps 4-8 may be based on incorrect assumptions.



## Scenario context

AI is a tool, not a religion. Use **Code** for deterministic logic (banking, math). Use **Traditional ML** for structured predictions (fraud, recommendations). Use **LLMs** for unstructured text (summarization, chat). Use **Agents** only when a task genuinely requires multi-step planning and external tool calls.

---

Choosing the right approach starts by analyzing your problem's input structure, output requirements…

This walkthrough focuses on **The Decision Flow: From Problem to Paradigm**.

Before making a design choice, clarify:

- Choosing the right approach starts by analyzing your problem's input structure, output requirements, and tolerance for non-determinism.

**Concepts in play:** llm-agents-and-langgraph, deep-learning-foundations, event-driven-architecture, hash-tables-and-bloom-filters

## Stage 1 — One Codebase, Four Paradigms: Know Before You Reach for the LLM

- The most expensive mistake in modern software is using an LLM for a problem deterministic code solves in 5 lines.
- Before adding an AI component, ask two questions: 1.
- Does the input have known structure?
- If both answers are no and the input is natural language, then LLMs are the right tool.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** If both answers are no and the input is natural language, then LLMs are the right tool.

## Stage 2 — Deep Dive: The Four Paradigms at a Glance

- A quick reference before diving deeper.
- Each paradigm has a distinct role — picking the wrong one costs time, money, and unnecessary complexity.
- Think of it as a filter: start at Code and only move down the list when the previous paradigm genuinely cannot handle the problem.

*Applies **deep-learning-foundations** to the decision above.*

**Observable outcome:** Think of it as a filter: start at Code and only move down the list when the previous paradigm genuinely cannot handle the problem.

## Stage 3 — Pure Code: When Determinism Is Non-Negotiable

- Any operation where you can write an explicit rule belongs here.
- When code beats AI: Banking transactions, data migrations, format validation, mathematical computations, protocol parsing.
- The rule: if a junior developer could write a test that covers every case, write code.

*Applies **event-driven-architecture** to the decision above.*

**Observable outcome:** The rule: if a junior developer could write a test that covers every case, write code.

## Stage 4 — Traditional ML: Patterns in Structured Tabular Data

- Use ML when the rule is too complex to write by hand, but the input is structured (rows and columns with known features).
- ML requires: labeled training data, feature engineering, model evaluation, and a retraining pipeline.
- If you don't have those, use rules instead.

*Applies **hash-tables-and-bloom-filters** to the decision above.*

```mermaid
flowchart LR
    Features[Structured Features age, amount, location] --> Model[ML Model XGBoost / Random Forest]
    Model --> Prediction[Score or Label fraud probability]
```

**Observable outcome:** If you don't have those, use rules instead.

## Stage 5 — Deep Dive: LLMs: When the Input Is Unstructured Text

- LLMs excel at tasks where the input is free-form text and the output is also text (or a structured schema derived from text).
- Cost reminder: Every LLM call costs money and adds latency.
- Never use an LLM for tasks that code or a simple ML model can solve.
- 🔬 Internals Traditional software follows deterministic control flow: the same input always produces the same output via explicit conditional branches.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** 🔬 Internals Traditional software follows deterministic control flow: the same input always produces the same output via explicit conditional branches.

## Stage 6 — Agents: For Multi-Step Goals That Require External Tools

- Use agents when completing the task requires: 1.
- Multiple actions (not just one generation) 2.
- Calling external APIs or tools (not just text transformation) 3.
- Adapting plans based on intermediate results Red flag: If you're describing your agent as "it just generates text and returns it," you needed a plain LLM call, not an agent.

*Applies **deep-learning-foundations** to the decision above.*

**Observable outcome:** Adapting plans based on intermediate results Red flag: If you're describing your agent as "it just generates text and returns it," you needed a plain LLM call, not an agent.

## End-to-end outcome

Choosing the right approach starts by analyzing your problem's input structure, output requirements, and tolerance for non-determinism.

## Alternatives considered

### Decision Guide: Choosing the Right Tool

The flowchart below turns the two intro questions into a repeatable process you can run mentally for any new backlog item.

### HuggingFace Transformers & scikit-learn: LLM Inference vs Traditional ML Pipeline Side by Side

HuggingFace Transformers is the go-to Python library for loading and running open-source LLMs (GPT-2, LLaMA, Mistral).


## Failure and degraded modes

### Trade-offs & Failure Modes: When Each Paradigm Breaks Down

- **Detection:** This flowchart maps any new requirement through a four-question decision tree that routes it to the right paradigm.
- **User or operator impact:** This flowchart maps any new requirement through a four-question decision tree that routes it to the right paradigm.



## Scenario context

Most bad LLM products do not fail because the model is weak. They fail because teams wrap a maybe-useful model in too much architecture: prompt spaghetti, no eval harness, weak tool schemas, huge context windows, agent chains nobody can explain, and zero cost ceilings. Use an LLM only when the problem is genuinely ambiguous, keep the architecture linear for as long as possible, and remove or simplify the model when…

This walkthrough focuses on **Why Over-Engineered LLM Systems Rot Faster Than Normal Integrations**.

Before making a design choice, clarify:

- The reason LLM architecture gets messy so quickly is that uncertainty can enter at several layers, and teams often treat every bad outcome as a prompting issue even when the real cause is orchestration.
- The Internals of an Over-Engineered LLM Request A single user message can fan out through multiple uncertain steps: 1.
- a router decides whether the request is search, chat, classification, or tool use 2.

**Concepts in play:** caching, llm-prompt-engineering, sorting-and-searching-algorithms, database-indexing

## Stage 1 — Four Ordinary LLM Projects That Drift Into Complexity

- The pattern usually starts with a perfectly reasonable idea.
- A support team wants a copilot that drafts replies.
- An internal platform team wants a search assistant over docs and runbooks.
- Ops wants a ticket triage bot.

*Applies **caching** to the decision above.*

**Observable outcome:** Ops wants a ticket triage bot.

## Stage 2 — The First Question: Do You Need Probabilistic Software at All?

- LLMs are strongest when the task is fuzzy: summarizing long text, rewriting for tone, extracting meaning from messy language, or generating a first draft from incomplete context.
- They are weakest when the job is bounded, rule-heavy, or directly tied to a source of truth your software already owns.
- That means you should actively try to route around the model .
- Use deterministic software first when: - the valid output space is small and known in advance - the answer must be exact, auditable, or regulation-sensitive - the task is really lookup plus formatting - a rules engine o…

*Applies **llm-prompt-engineering** to the decision above.*

**Observable outcome:** Use deterministic software first when: - the valid output space is small and known in advance - the answer must be exact, auditable, or regulation-sensitive - the task is really lookup plus formatting - a rules engine o…

## Stage 3 — How Prompt Spaghetti and Architecture Inflation Happen

- Most production LLM failures are software design failures wearing AI clothes.
- The model becomes the excuse for avoiding boundaries, schemas, tests, and fallback paths that normal engineering would demand.
- Prompt spaghetti is just hidden business logic If your application behavior depends on seven prompt variants, two invisible system messages, and a few "temporary" string concatenations spread across the codebase, you do…
- You have undocumented business logic.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```architecture
{
  "version": 1,
  "title": "How Prompt Spaghetti and Architecture Inflation Happen - Subsystem Overview",
  "description": "Decoupled system architecture isolating ingress, stateless logic, and persistent datastores.",
  "view": "high-level",
  "pattern": "microservices",
  "direction": "LR",
  "groups": [
    { "id": "edge", "label": "Edge Traffic Ingress", "tone": "slate" },
    { "id": "services", "label": "Stateless Microservices", "tone": "blue" },
    { "id": "data", "label": "Datastore Infrastructure", "tone": "green" }
  ],
  "nodes": [
    { "id": "gw", "component": "api-gateway", "label": "API Gateway", "group": "edge", "badge": "TLS & Auth" },
    { "id": "app", "component": "workflow-service", "label": "Domain Service", "group": "services", "badge": "Stateless" },
    { "id": "cache", "component": "redis", "label": "Redis Cache", "group": "data", "badge": "Sub-ms Reads" },
    { "id": "db", "component": "postgres", "label": "Transactional DB", "group": "data", "badge": "ACID Log" }
  ],
  "edges": [
    { "from": "gw", "to": "app", "kind": "request", "label": "gRPC Request" },
    { "from": "app", "to": "cache", "kind": "read", "label": "Cache Lookups" },
    { "from": "app", "to": "db", "kind": "write", "label": "Durable Write" }
  ],
  "takeaway": {
    "title": "Architecture Boundary Invariant",
    "body": "Stateless compute nodes scale horizontally while shielding underlying database clusters behind fast cache layers."
  }
}
```

**Observable outcome:** You have undocumented business logic.

## Stage 4 — A Keep-Simplify-Remove Flow for LLM Features

- Before you optimize prompts, decide whether the model should stay in the path at all.
- The diagram below is intentionally simple: it treats the LLM as one component in a product system, not the center of the universe.
- Read it from top to bottom and notice how often the safe answer is "use normal software first." This flow is also a debugging aid.
- If your design jumps straight from "incoming task" to "agent planner," you skipped the most important engineering question.

*Applies **database-indexing** to the decision above.*

**Observable outcome:** If your design jumps straight from "incoming task" to "agent planner," you skipped the most important engineering question.

## Stage 5 — Five Small Python Guards That Make an LLM Feature Behave More Like Software

- These examples are intentionally small and runnable.
- They are not framework demos; they are engineering habits you can drop into a prototype today.
- Read them as guardrails for pruning complexity, not as reasons to add more layers.
- Example 1: Route Around the Model When the Task Is Bounded If this kind of router handles a large share of traffic, that is good news.

*Applies **caching** to the decision above.*

**Observable outcome:** Example 1: Route Around the Model When the Task Is Bounded If this kind of router handles a large share of traffic, that is good news.

## Stage 6 — Pydantic: How Strict Schemas Stop Tool and Output Drift in Practice

- Pydantic is an open-source Python validation framework that turns loose JSON into explicit contracts.
- In LLM systems, that matters because structured output is where "looks correct" often becomes "silently wrong." If the model says a ticket severity is "urgent-ish" when your workflow expects low medium high , you want a…
- This snippet keeps the model on a short leash.
- It validates category, urgency, and confidence before the rest of the application trusts the output.

*Applies **llm-prompt-engineering** to the decision above.*

**Observable outcome:** It validates category, urgency, and confidence before the rest of the application trusts the output.

## End-to-end outcome

After working through the stages, you should be able to explain:

- - The best LLM architecture is usually the smallest one that survives contact with production.
- - If a deterministic path handles the request well enough, that is not a compromise.

## Alternatives considered

### When Is It Too Much? A Decision Guide for Simplifying or Removing the LLM

Here is the blunt answer: it is too much when the LLM is no longer the smallest useful source of intelligence in the system.


## Failure and degraded modes

### The anti-patterns above become easier to see when you look at common product shapes instead of abstract architecture diagrams.

- **Detection:** The anti-patterns above become easier to see when you look at common product shapes instead of abstract architecture diagrams.
- **User or operator impact:** The anti-patterns above become easier to see when you look at common product shapes instead of abstract architecture diagrams.

### Case Study: Support Copilot

- **Detection:** Input: a customer asks for a refund after a delayed shipment.
- **User or operator impact:** Input: a customer asks for a refund after a delayed shipment.

### Case Study: Internal Search Assistant

- **Detection:** Input: an engineer asks, "How do I rotate Kafka credentials in staging?" Better process: search docs, rank results, summarize the top two, and include links.
- **User or operator impact:** Input: an engineer asks, "How do I rotate Kafka credentials in staging?" Better process: search docs, rank results, summarize the top two, and include links.

### Case Study: Ticket Triage Bot

- **Detection:** Input: a bug report arrives with messy text, logs, and screenshots.
- **User or operator impact:** Input: a bug report arrives with messy text, logs, and screenshots.

### The Trade-Offs That Quietly Turn an LLM Feature Into a Liability

- **Detection:** Every LLM feature carries hidden trade-offs, but some are especially easy to ignore during prototype week.
- **User or operator impact:** Every LLM feature carries hidden trade-offs, but some are especially easy to ignore during prototype week.



Start with when agents make sense and how tool use changes LLM applications.

Agent Foundations

## Scenario context

Prompt Engineering is the art of writing instructions that guide an LLM toward the answer you want. Zero-Shot, Few-Shot, and Chain-of-Thought are systematic techniques — not guesswork — that can dramatically improve accuracy without changing a single model weight.

---

Research shows LLMs pay the most attention to content at the **beginning** and **end** of the prompt. Information in the middle is often skipped.

`…

This walkthrough focuses on **The "Lost in the Middle" Problem**.

Before making a design choice, clarify:

- Research shows LLMs pay the most attention to content at the beginning and end of the prompt.
- Information in the middle is often skipped.
- Practical rule: Put your most important instructions or the specific text to analyze at either the very start or the very end.

**Concepts in play:** llm-prompt-engineering, consistency-replication, tree-data-structures, deep-learning-foundations

## Stage 1 — The Super-Smart Intern With Zero Initiative

- Imagine a brilliant intern who has read every book ever written but has no common sense.
- Give them a vague instruction ("Fix this") and they stare blankly.
- Give them a structured instruction and they execute perfectly.
- Prompt Engineering is the art of writing the structured instruction.

*Applies **llm-prompt-engineering** to the decision above.*

**Observable outcome:** Prompt Engineering is the art of writing the structured instruction.

## Stage 2 — How Prompts Work: The Fundamentals

- Every time you send text to an LLM, the entire input — system message, instructions, examples, and your question — is packed into a context window .
- The model reads it all at once and predicts the next most likely token.
- That prediction is directly shaped by every word you included.
- Three core ingredients of a well-formed prompt: Missing any one of these is the most common source of vague or off-topic responses.

*Applies **consistency-replication** to the decision above.*

**Observable outcome:** Three core ingredients of a well-formed prompt: Missing any one of these is the most common source of vague or off-topic responses.

## Stage 3 — The Four Core Techniques

- Zero-Shot: Ask Directly Give the task with no examples: Best when: The task is well-defined and the model likely saw similar tasks in pre-training.
- Few-Shot: Teach by Example Provide 2–5 examples before the actual question: Why it works: The examples prime the model's "in-context learning." It extracts the pattern from the examples rather than relying on training d…
- Tip: Select diverse examples that cover edge cases.
- Avoid examples biased toward one label.

*Applies **tree-data-structures** to the decision above.*

```
Classify this movie review as Positive or Negative:
"The acting was incredible and the plot kept me on the edge of my seat."

Classification:
```

```
Review: "Disappointing. Slow pacing, weak script." → Negative
Review: "A masterpiece. Emotional and visually stunning." → Positive
Review: "Predictable but fun summer blockbuster." → ?
```

**Observable outcome:** Avoid examples biased toward one label.

## Stage 4 — Deep Dive: Self-Consistency and Tree-of-Thought

- Self-Consistency Instead of one CoT sample, generate 3–5 different reasoning chains and vote on the most common answer.
- Reduces reasoning variance on math problems.
- Tree-of-Thought (ToT) For planning problems, generate multiple "thought branches" at each step, evaluate which branch is most promising, and expand only that branch.
- Outperforms linear CoT on reasoning tasks that require backtracking.

*Applies **deep-learning-foundations** to the decision above.*

```python
answers = [generate_cot_response(prompt) for _ in range(5)]
from collections import Counter
final = Counter(answers).most_common(1)[0][0]
```

**Observable outcome:** Outperforms linear CoT on reasoning tasks that require backtracking.

## Stage 5 — Real-World Applications: Prompt Engineering in Production

- Prompt engineering is not just a prototyping trick — it underpins real product features across industries.
- The table below shows common industry use cases and which technique typically drives the best results.
- Case Study — SaaS Customer Triage: A mid-size SaaS company was spending 8 engineer-hours per week manually labeling incoming support tickets before routing them to the correct team.
- After building a Few-Shot classifier prompt with 4 labeled examples per category (billing, bug, feature request, account access), the model achieved 91% routing accuracy on a held-out test set.

*Applies **llm-prompt-engineering** to the decision above.*

```mermaid
flowchart TD
    Task[Define Generation Task]
    WellDefined{Task clearly defined?}
    Examples{Have labeled examples?}
    MultiStep{Requires multi-step reasoning?}
    Variance{High-stakes, reduce variance?}

    ZeroShot["Zero-Shot (simple, direct)"]
    FewShot["Few-Shot (2-5 examples)"]
    CoT["Chain-of-Thought (step-by-step reasoning)"]
    SelfConsist["Self-Consistency (vote on 5 CoT runs)"]
    ReAct["ReAct (reason + act + observe)"]

    Task --> WellDefined
    WellDefined -->|No: rephrase| Task
    WellDefined -->|Yes| Examples
    Examples -->|No| ZeroShot
    Examples -->|Yes| FewShot
    FewShot --> MultiStep
    MultiStep -->|Yes| CoT
    MultiStep -->|No| FewShot
    CoT --> Variance
    Variance -->|Yes| SelfConsist
    Variance -->|No + tools| ReAct
```

**Observable outcome:** After building a Few-Shot classifier prompt with 4 labeled examples per category (billing, bug, feature request, account access), the model achieved 91% routing accuracy on a held-out test set.

## Stage 6 — Hands-On Exercises: Putting the Techniques to Work

- The fastest way to internalize prompt engineering is to rewrite bad prompts.
- These three exercises cover the three most common failure modes.
- Exercise 1 — Zero-Shot to Few-Shot upgrade You have this Zero-Shot prompt that struggles with mixed-sentiment reviews: Rewrite it as a Few-Shot prompt with 3 examples that cover each label class.
- Make sure one example is genuinely mixed-sentiment so the model learns the boundary.

*Applies **consistency-replication** to the decision above.*

```
Classify this review as Positive, Negative, or Mixed:
"The camera is stunning but battery life is terrible."
```

```
A jacket costs $120. It's 30% off. What is the final price?
```

**Observable outcome:** Make sure one example is genuinely mixed-sentiment so the model learns the boundary.

## End-to-end outcome

Research shows LLMs pay the most attention to content at the beginning and end of the prompt.

## Alternatives considered

### Decision Guide: Prompting Technique Selection

Not sure which technique to reach for?


## Failure and degraded modes

### Trade-offs & Failure Modes: Common Anti-Patterns

- **Detection:** Trade-offs & Failure Modes: Common Anti-Patterns
- **User or operator impact:** Trade-offs & Failure Modes: Common Anti-Patterns



## Scenario context

LangChain's LCEL pipe operator (`|`) wires prompts, models, and output parsers into composable chains — swap OpenAI for Anthropic or Ollama by changing one line without touching the rest of your code.

---

Imagine you are three days into building a customer support bot. You hardcoded the OpenAI API, hand-rolled the request payload, parsed the JSON response yourself, and added a retry loop for rate limits. It works.…

This walkthrough focuses on **One LLM API Today, Rewrite Tomorrow: The Customer Support Bot Problem**.

Before making a design choice, clarify:

- Imagine you are three days into building a customer support bot.
- You hardcoded the OpenAI API, hand-rolled the request payload, parsed the JSON response yourself, and added a retry loop for rate limits.
- Then your manager asks you to benchmark GPT-4o against Claude 3.5 Sonnet.

**Concepts in play:** llm-agents-and-langgraph, deep-learning-foundations, api-gateway, rate-limiting

## Stage 1 — LangChain's Core Primitives: Prompts, Models, and Parsers

- Every LangChain pipeline — no matter how complex — is built from three building blocks.
- Understanding these three objects is 80% of what you need to be productive with LangChain.
- ChatPromptTemplate and the {variable} Syntax A ChatPromptTemplate is a reusable message template with named placeholders.
- You define it once with {curly-brace} variables, then call .invoke({"variable": "value"}) to fill them in at runtime.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that explains topics clearly."),
    ("human", "Explain {topic} in simple terms for a {audience} audience."),
])

# Fill in the variables
filled = prompt.invoke({"topic": "neural networks", "audience": "high school student"})
print(filled.messages)
# [SystemMessage(content='You are a helpful assistant...'),
#  HumanMessage(content='Explain neural networks in simple terms for a high school student audience.')]
```

```python
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="You are a concise technical writer."),
    HumanMessage(content="What is a transformer model?"),
    AIMessage(content="A transformer is a neural network that uses attention mechanisms..."),
    HumanMessage(content="Can you give a one-sentence summary?"),
]
```

**Observable outcome:** You define it once with {curly-brace} variables, then call .invoke({"variable": "value"}) to fill them in at runtime.

## Stage 2 — LCEL: How the Pipe Operator Assembles a Chain

- LCEL (LangChain Expression Language) is the modern way to compose LangChain components.
- Its key innovation is a single operator: the pipe ( ).
- You have already seen all three pieces — prompt, model, and parser.
- LCEL lets you connect them left to right: That one line creates a fully functional pipeline.

*Applies **deep-learning-foundations** to the decision above.*

```python
chain = prompt | llm | parser
```

```python
# pip install langchain langchain-openai
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful technical explainer."),
    ("human", "Explain {topic} in one paragraph for a {level} audience."),
])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({"topic": "REST APIs", "level": "beginner"})
print(result)
```

**Observable outcome:** LCEL lets you connect them left to right: That one line creates a fully functional pipeline.

## Stage 3 — Visualizing the LCEL Data Flow

- The pipe operator is easy to read left-to-right, but it helps to see what data type flows between each step.
- Here is the complete transformation: Notice that each arrow represents a type transformation : a dict becomes a list of messages, messages become an AIMessage , and AIMessage becomes a str .
- This is why the Runnable protocol is powerful — each component only needs to know its input and output type, not anything about its neighbors.

*Applies **api-gateway** to the decision above.*

```mermaid
graph TD
    A[Input Dictionary {topic: '...', level: '...'}] --> B["ChatPromptTemplate (fills variables)"]
    B --> C[List of ChatMessages [SystemMessage, HumanMessage]]
    C --> D["ChatOpenAI / ChatAnthropic / ChatOllama (calls LLM API)"]
    D --> E["AIMessage (raw model response object)"]
    E --> F["StrOutputParser / JsonOutputParser (extracts content)"]
    F --> G["Plain str or dict (your application data)"]

    style A fill:#e8f4f8,stroke:#2196F3
    style D fill:#fff3e0,stroke:#FF9800
    style G fill:#e8f5e9,stroke:#4CAF50
```

**Observable outcome:** This is why the Runnable protocol is powerful — each component only needs to know its input and output type, not anything about its neighbors.

## Stage 4 — Real-World Applications: LLM-Agnostic Deployment from Prototype to Production

- The single biggest benefit of LCEL is that swapping a model is a one-line change.
- The prompt, the parser, and all your business logic stay identical.
- Only the object assigned to llm changes.
- Here is the same explain topic function running against three different providers without a single structural change: All three produce a well-formed plain-text answer.

*Applies **rate-limiting** to the decision above.*

```python
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical explainer. Keep answers to 3 sentences max."),
    ("human", "Explain {topic} to a {level} developer."),
])
parser = StrOutputParser()

def explain_topic(topic: str, level: str, llm) -> str:
    chain = prompt | llm | parser
    return chain.invoke({"topic": topic, "level": level})

# ── OpenAI ──────────────────────────────────────────────────────
openai_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
print("OpenAI:", explain_topic("LCEL", "beginner", openai_llm))

# ── Anthropic ───────────────────────────────────────────────────
anthropic_llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0)
print("Anthropic:", explain_topic("LCEL", "beginner", anthropic_llm))

# ── Ollama (local — no API key required) ────────────────────────
# Run: ollama pull llama3.2 first
ollama_llm = ChatOllama(model="llama3.2", temperature=0)
print("Ollama:", explain_topic("LCEL", "beginner", ollama_llm))
```

**Observable outcome:** Here is the same explain topic function running against three different providers without a single structural change: All three produce a well-formed plain-text answer.

## Stage 5 — Deep Dive: How LCEL's Runnable Protocol Wires Components Together

- Every component you have used — ChatPromptTemplate , ChatOpenAI , StrOutputParser — implements LangChain's Runnable protocol.
- This single interface contract is what makes the pipe operator work.
- A Runnable exposes four methods, and you get all of them for free on every LCEL chain: When you call chain.invoke({"topic": "..."}) , LangChain walks the pipe left to right — it calls .invoke() on the prompt, passes the…
- The same walk applies to .stream() , meaning streaming is built into every LCEL chain without any extra plumbing: This uniformity also means you can inject custom Python logic anywhere in the chain using RunnableLambda(…

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
# Token-by-token streaming — no custom iterator needed
for chunk in chain.stream({"topic": "LCEL", "level": "beginner"}):
    print(chunk, end="", flush=True)
```

**Observable outcome:** The same walk applies to .stream() , meaning streaming is built into every LCEL chain without any extra plumbing: This uniformity also means you can inject custom Python logic anywhere in the chain using RunnableLambda(…

## Stage 6 — LangChain's Open-Source Architecture: Installation, LCEL, and the Runnable Protocol

- LangChain (github.com/langchain-ai/langchain) is an open-source Python and TypeScript framework maintained by LangChain, Inc.
- It is built around a layered package structure that keeps the core abstractions separate from provider-specific integrations: - langchain-core — The foundational Runnable protocol, base classes for prompts, models, pars…
- This package is the only dependency shared by everything.
- Installed automatically when you install langchain .

*Applies **deep-learning-foundations** to the decision above.*

```architecture
{
  "version": 1,
  "title": "LangChain's Open-Source Architecture: Installation, LCEL, and the Runnable Protocol - Subsystem Overview",
  "description": "Decoupled system architecture isolating ingress, stateless logic, and persistent datastores.",
  "view": "high-level",
  "pattern": "microservices",
  "direction": "LR",
  "groups": [
    { "id": "edge", "label": "Edge Traffic Ingress", "tone": "slate" },
    { "id": "services", "label": "Stateless Microservices", "tone": "blue" },
    { "id": "data", "label": "Datastore Infrastructure", "tone": "green" }
  ],
  "nodes": [
    { "id": "gw", "component": "api-gateway", "label": "API Gateway", "group": "edge", "badge": "TLS & Auth" },
    { "id": "app", "component": "workflow-service", "label": "Domain Service", "group": "services", "badge": "Stateless" },
    { "id": "cache", "component": "redis", "label": "Redis Cache", "group": "data", "badge": "Sub-ms Reads" },
    { "id": "db", "component": "postgres", "label": "Transactional DB", "group": "data", "badge": "ACID Log" }
  ],
  "edges": [
    { "from": "gw", "to": "app", "kind": "request", "label": "gRPC Request" },
    { "from": "app", "to": "cache", "kind": "read", "label": "Cache Lookups" },
    { "from": "app", "to": "db", "kind": "write", "label": "Durable Write" }
  ],
  "takeaway": {
    "title": "Architecture Boundary Invariant",
    "body": "Stateless compute nodes scale horizontally while shielding underlying database clusters behind fast cache layers."
  }
}
```

```bash
# Core framework
pip install langchain langchain-core

# Provider integrations
pip install langchain-openai      # OpenAI and Azure OpenAI
pip install langchain-anthropic   # Anthropic Claude
pip install langchain-ollama      # Ollama local models

# Optional: for JSON parsing utilities
pip install langchain-community
```

**Observable outcome:** Installed automatically when you install langchain .

## End-to-end outcome

Imagine you are three days into building a customer support bot.

## Alternatives considered

### Decision Guide: LangChain vs. Direct API Calls

Decision Guide: LangChain vs. Direct API Calls


## Failure and degraded modes

### Trade-offs and Failure Modes: What LangChain Does Not Fix

- **Detection:** LangChain's abstractions deliver genuine value, but they introduce their own costs.
- **User or operator impact:** LangChain's abstractions deliver genuine value, but they introduce their own costs.



## Scenario context

LLMs are stateless — every API call starts fresh. LangChain memory classes (Buffer, Window, Summary, SummaryBuffer) explicitly inject history into each call, and `RunnableWithMessageHistory` is the modern LCEL replacement for the legacy `ConversationChain`.
---

You have built a customer support chatbot. The first message goes perfectly:

> **User:** Hi, my account number is **ACC-8837** and I cannot log in.
> **Bot…

This walkthrough focuses on **The Amnesia Tax: Every Conversation Starts from Zero**.

Before making a design choice, clarify:

- You have built a customer support chatbot.
- The first message goes perfectly: User: Hi, my account number is ACC-8837 and I cannot log in.
- Account ACC-8837 is locked after three failed attempts.

**Concepts in play:** llm-agents-and-langgraph, python-concurrency-nogil, caching, consistency-replication

## Stage 1 — What the LLM API Actually Receives on Every Call

- Before choosing a memory strategy, it helps to see exactly what the API receives.
- Every call to llm.invoke(messages) sends a structured list of messages.
- Once the response is returned, the API discards everything.
- There is no session, no thread ID, no continuation token.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke([HumanMessage(content="My account is ACC-8837.")])
# The model processes this, returns a response, and forgets everything.
```

```python
from langchain_core.messages import HumanMessage, AIMessage

messages = [
    HumanMessage(content="My account number is ACC-8837."),
    AIMessage(content="Got it. Account ACC-8837 is locked."),
    HumanMessage(content="Can you resend the reset link?"),
]
response = llm.invoke(messages)
# The model can now reference ACC-8837 because the history is in THIS call.
```

**Observable outcome:** There is no session, no thread ID, no continuation token.

## Stage 2 — Memory Flow: From User Input to LLM Response

- Every memory-backed chain follows a consistent read-before-write cycle.
- The memory object wraps each LLM call like a decorator: it injects history before the model sees the prompt, and it persists the exchange after the response arrives.
- Figure: The memory read-before-write cycle on every conversation turn.
- The LLM itself remains stateless; the history management lives entirely in the LangChain chain layer above it.

*Applies **python-concurrency-nogil** to the decision above.*

```mermaid
graph TD
    A[User sends a message] --> B[Chain receives input]
    B --> C[Memory.load_memory_variables]
    C --> D[Retrieve stored history or running summary]
    D --> E[Assemble full prompt: history + new message]
    E --> F[LLM API call with full context]
    F --> G[LLM generates response]
    G --> H[Memory.save_context]
    H --> I[Write new input + output to memory store]
    I --> J[Return response to user]
    J --> A
```

**Observable outcome:** The LLM itself remains stateless; the history management lives entirely in the LangChain chain layer above it.

## Stage 3 — Deep Dive: Inside LangChain's Memory Architecture

- Under the Internals: How Memory Classes Store and Retrieve State All four LangChain memory classes share the same two-method interface: load memory variables(inputs) and save context(inputs, outputs) .
- The ConversationChain (and RunnableWithMessageHistory ) calls them in sequence around every LLM invocation.
- Internally, every class wraps a ChatMessageHistory object — a simple ordered list of BaseMessage objects ( HumanMessage , AIMessage , SystemMessage ).
- The classes differ only in what they do before handing that list to the prompt: - BufferMemory returns the full list unchanged.

*Applies **caching** to the decision above.*

```architecture
{
  "version": 1,
  "title": "Deep Dive: Inside LangChain's Memory Architecture - Subsystem Overview",
  "description": "Decoupled system architecture isolating ingress, stateless logic, and persistent datastores.",
  "view": "high-level",
  "pattern": "microservices",
  "direction": "LR",
  "groups": [
    { "id": "edge", "label": "Edge Traffic Ingress", "tone": "slate" },
    { "id": "services", "label": "Stateless Microservices", "tone": "blue" },
    { "id": "data", "label": "Datastore Infrastructure", "tone": "green" }
  ],
  "nodes": [
    { "id": "gw", "component": "api-gateway", "label": "API Gateway", "group": "edge", "badge": "TLS & Auth" },
    { "id": "app", "component": "workflow-service", "label": "Domain Service", "group": "services", "badge": "Stateless" },
    { "id": "cache", "component": "redis", "label": "Redis Cache", "group": "data", "badge": "Sub-ms Reads" },
    { "id": "db", "component": "postgres", "label": "Transactional DB", "group": "data", "badge": "ACID Log" }
  ],
  "edges": [
    { "from": "gw", "to": "app", "kind": "request", "label": "gRPC Request" },
    { "from": "app", "to": "cache", "kind": "read", "label": "Cache Lookups" },
    { "from": "app", "to": "db", "kind": "write", "label": "Durable Write" }
  ],
  "takeaway": {
    "title": "Architecture Boundary Invariant",
    "body": "Stateless compute nodes scale horizontally while shielding underlying database clusters behind fast cache layers."
  }
}
```

```
Progressively summarize the lines of conversation provided, adding onto
the previous summary and returning a new summary.

SUMMARY:
{summary}

NEW LINES OF CONVERSATION:
{new_lines}

NEW SUMMARY:
```

**Observable outcome:** The classes differ only in what they do before handing that list to the prompt: - BufferMemory returns the full list unchanged.

## Stage 4 — Real-World Application: A Multi-Turn Code Review Assistant

- A Code Review Assistant is an ideal showcase for memory: the user submits code in the first turn and then asks follow-up questions that only make sense in the context of what was previously reviewed.
- Without memory, every follow-up requires the user to paste the code again.
- The following five-turn session uses ConversationSummaryBufferMemory .
- The first three turns fit comfortably within the 800-token verbatim buffer; by turn four, older context is summarized, yet the model retains the critical facts.

*Applies **consistency-replication** to the decision above.*

```python
from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=800)

template = """You are a senior code reviewer. Be concise and specific.

{history}
Human: {input}
Assistant:"""

prompt = PromptTemplate(input_variables=["history", "input"], template=template)
chain = ConversationChain(llm=llm, memory=memory, prompt=prompt)

# Turn 1 — user submits code for review
response = chain.predict(input="""
Review this Python function:
def fetch_user(user_id):
    conn = db.connect()
    result = conn.execute(f"SELECT * FROM users WHERE id = {user_id}")
    return result.fetchone()
""")
# → Critical issue: SQL injection vulnerability on the f-string query.
#   Use a parameterized query: conn.execute("SELECT * FROM users WHERE id = ?", (user_id,))
#   Also: the connection is never closed — wrap in a context manager.

# Turn 2 — follow-up: ask for corrected code
response = chain.predict(input="Show me the corrected version with both fixes.")
# → def fetch_user(user_id):
#       with db.connect() as conn:
#           result = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,))
#           return result.fetchone()

# Turn 3 — ask about a related pattern
response = chain.predict(input="Is SQLAlchemy connection pooling better here?")
# → Yes — use create_engine() with pool_size and pool_recycle for production.
#   The context manager approach you have now is fine for scripts.

# Turn 4 — reference the corrected code from turn 2
response = chain.predict(input="Can the function you showed handle None as user_id?")
# → No. Add a guard: if user_id is None: raise ValueError("user_id must not be None")
#   The parameterized query would pass NULL to the database, returning unexpected results.

# Turn 5 — synthesize all findings
response = chain.predict(input="Summarize every issue found and the final recommended code.")
# → Three issues addressed:
#   1. SQL injection — fixed with parameterized query
#   2. Unclosed connection — fixed with context manager
#   3. None input — add a ValueError guard before the query
print(response)
```

**Observable outcome:** The first three turns fit comfortably within the 800-token verbatim buffer; by turn four, older context is summarized, yet the model retains the critical facts.

## Stage 5 — Community Memory Backends: Redis, MongoDB, and DynamoDB

- The in-memory ChatMessageHistory is appropriate for single-process development and testing.
- In production, where multiple workers serve the same user concurrently or sessions must survive process restarts, you need a persistent backend.
- The langchain-community package ships ready-made history implementations for the most common stores.
- They all implement BaseChatMessageHistory and can be dropped in as the chat memory argument of any LangChain memory class.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
# pip install langchain-community redis
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

history = RedisChatMessageHistory(
    session_id="user-ACC-8837",
    url="redis://localhost:6379",
    ttl=3600,  # Expire inactive sessions after 1 hour
)
memory = ConversationSummaryBufferMemory(
    llm=llm,
    chat_memory=history,
    max_token_limit=1000,
    return_messages=True,
)
```

```python
# MongoDB backend
from langchain_community.chat_message_histories import MongoDBChatMessageHistory

history = MongoDBChatMessageHistory(
    session_id="user-ACC-8837",
    connection_string="mongodb://localhost:27017",
    database_name="chat_sessions",
    collection_name="message_history",
)
```

**Observable outcome:** They all implement BaseChatMessageHistory and can be dropped in as the chat memory argument of any LangChain memory class.

## Stage 6 — The Modern LCEL Memory Pattern: RunnableWithMessageHistory

- ConversationChain is the legacy API.
- For any new project, use RunnableWithMessageHistory — LangChain's LCEL-native memory wrapper.
- It is composable with the pipe operator, supports async streaming natively, and cleanly separates session management (your responsibility) from chain logic (LangChain's responsibility).
- To persist sessions across restarts, replace ChatMessageHistory() with RedisChatMessageHistory(session id=session id, url=REDIS URL) inside get session history .

*Applies **python-concurrency-nogil** to the decision above.*

```python
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful coding assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | llm  # Clean LCEL pipe — no ConversationChain wrapper

# In-memory store: swap for RedisChatMessageHistory in production
store: dict[str, BaseChatMessageHistory] = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# Pass session identity via config on every call
config = {"configurable": {"session_id": "user-ACC-8837"}}

r1 = chain_with_history.invoke({"input": "My account is ACC-8837."}, config=config)
r2 = chain_with_history.invoke({"input": "What account did I just mention?"}, config=config)
print(r2.content)
# → "You mentioned account ACC-8837."
```

**Observable outcome:** To persist sessions across restarts, replace ChatMessageHistory() with RedisChatMessageHistory(session id=session id, url=REDIS URL) inside get session history .

## End-to-end outcome

After working through the stages, you should be able to explain:

- ConversationBufferMemory — Append Everything The simplest approach: store every message in a list and send the complete list on every call.
- The memory object implements two key methods on every turn: load memory variables({}) reads the history before the LLM call, and save context({"input": ...}, {"output": ...}) writes the new exchange back after.
- Set a token budget before your first deploy, not after your first incident.
- ConversationBufferMemory is tempting because it requires zero configuration.

## Alternatives considered

### Decision Guide: Choosing the Right Memory Strategy

Decision Guide: Choosing the Right Memory Strategy


## Failure and degraded modes

### Trade-offs & Failure Modes in Conversational Memory

- **Detection:** The most expensive lesson teams learn is deploying ConversationBufferMemory to production and discovering that power users with 150-turn sessions start hitting context window limits at 2 AM.
- **User or operator impact:** The most expensive lesson teams learn is deploying ConversationBufferMemory to production and discovering that power users with 150-turn sessions start hitting context window limits at 2 AM.



## Scenario context

Consider this prompt to a raw language model:

> *"What's the current weather in Tokyo, and how does today's high compare to yesterday's?"*

The model answers confidently. It might even produce a plausible-sounding temperature. But that answer is fabricated — the model's weights were frozen at training cutoff and it has no mechanism to observe the world at inference time. It cannot make an HTTP request. It cannot re…

This walkthrough focuses on **The Gap Between "Saying" and "Doing": Why LLMs Need Tools**.

Before making a design choice, clarify:

- Consider this prompt to a raw language model: "What's the current weather in Tokyo, and how does today's high compare to yesterday's?" The model answers confidently.
- It might even produce a plausible-sounding temperature.
- But that answer is fabricated — the model's weights were frozen at training cutoff and it has no mechanism to observe the world at inference time.

**Concepts in play:** llm-agents-and-langgraph, low-level-design-systems, llm-prompt-engineering, event-driven-architecture

## Stage 1 — What a Tool Actually Is: A Python Function the LLM Can Request

- A tool, in LangChain's model, is nothing exotic.
- It is a regular Python function that: 1.
- Has a name the LLM uses to refer to it.
- Has a description (from the docstring) that the LLM reads to decide when to call it.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Has a description (from the docstring) that the LLM reads to decide when to call it.

## Stage 2 — Turning Python Functions into LLM Tools with the @tool Decorator

- The Decorator Approach The @tool decorator from langchain core.tools is the simplest way to define a tool.
- Decorate a function, write a meaningful docstring, and annotate the parameters — that's it.
- The docstring is not documentation for human readers — it is the tool's entire identity to the LLM.
- The model reads it to decide whether to call the tool at all.

*Applies **low-level-design-systems** to the decision above.*

```python
# NOTE: External API calls below are mocked for local development.
# Replace the mock return values with real API calls in production.

from langchain_core.tools import tool

@tool
def get_current_weather(city: str) -> str:
    """
    Fetch the current weather for a given city.
    Returns a plain-text description including temperature and conditions.
    Use this when the user asks about current or today's weather anywhere.
    """
    # Mock: in production, call OpenWeatherMap or similar
    mock_data = {
        "tokyo": "Tokyo: 18°C, partly cloudy, humidity 65%",
        "london": "London: 12°C, overcast, humidity 80%",
        "new york": "New York: 22°C, sunny, humidity 45%",
    }
    return mock_data.get(city.lower(), f"Weather data unavailable for {city}")

@tool
def get_yesterday_weather(city: str) -> str:
    """
    Fetch yesterday's high temperature for a given city.
    Returns a plain-text description of yesterday's weather summary.
    Use this when the user wants to compare today's weather to yesterday's.
    """
    # Mock: in production, query a historical weather API
    mock_data = {
        "tokyo": "Tokyo yesterday: high of 15°C, light rain",
        "london": "London yesterday: high of 10°C, cloudy",
        "new york": "New York yesterday: high of 20°C, partly cloudy",
    }
    return mock_data.get(city.lower(), f"Historical weather unavailable for {city}")
```

```python
from pydantic import BaseModel, Field
from langchain_core.tools import tool

class StockLookupInput(BaseModel):
    ticker: str = Field(description="Stock ticker symbol, e.g. AAPL, TSLA, NVDA")
    currency: str = Field(default="USD", description="Currency for the price output (USD or EUR)")

@tool("get_stock_price", args_schema=StockLookupInput)
def get_stock_price(ticker: str, currency: str = "USD") -> str:
    """
    Return the current stock price for a given ticker symbol.
    Use this when the user asks about stock prices or market data.
    Do NOT use this for cryptocurrency prices.
    """
    # Mock: in production, call Yahoo Finance, Alpha Vantage, or Polygon.io
    mock_prices = {"AAPL": 213.07, "TSLA": 172.30, "NVDA": 118.42, "MSFT": 415.20}
    price = mock_prices.get(ticker.upper())
    if price is None:
        return f"No price data found for {ticker}"
    if currency == "EUR":
        price = round(price * 0.92, 2)
    return f"{ticker.upper()} is trading at {currency} {price:.2f}"
```

**Observable outcome:** The model reads it to decide whether to call the tool at all.

## Stage 3 — Deep Dive: How AgentExecutor Orchestrates the ReAct Loop

- The Internals: AgentExecutor as a Configurable Runnable Loop AgentExecutor is LangChain's agent runner.
- Internally, it is a while-loop wrapped around a runnable chain: on each iteration, it calls the LLM, inspects the output, decides whether to invoke a tool, runs the tool if needed, appends the observation to the message…
- It exits when the LLM produces a final answer (no more tool calls) or when a configured maximum step count is reached.
- The loop follows the ReAct (Reasoning + Acting) pattern, published by Yao et al.

*Applies **llm-prompt-engineering** to the decision above.*

```python
from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI

# Pull the standard ReAct prompt from LangChain Hub
# It instructs the LLM to emit Thought/Action/Observation cycles
react_prompt = hub.pull("hwchase17/react")

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_current_weather, get_yesterday_weather, get_stock_price]

agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,          # print the full reasoning trace
    max_iterations=6,      # hard stop to prevent runaway loops
    handle_parsing_errors=True,  # recover from malformed tool calls
)

result = executor.invoke({"input": "What's the current weather in Tokyo and how does it compare to yesterday?"})
print(result["output"])
```

```
> Entering new AgentExecutor chain...
Thought: I need to fetch the current weather in Tokyo and yesterday's weather to compare them.
Action: get_current_weather
Action Input: {"city": "tokyo"}
Observation: Tokyo: 18°C, partly cloudy, humidity 65%

Thought: Now I need yesterday's weather to compare.
Action: get_yesterday_weather
Action Input: {"city": "tokyo"}
Observation: Tokyo yesterday: high of 15°C, light rain

Thought: I have both data points. Today is 18°C and partly cloudy; yesterday peaked at 15°C with rain.
Final Answer: Tokyo is currently 18°C and partly cloudy — 3 degrees warmer than yesterday's high of 15°C, and the rain has cleared.
```

**Observable outcome:** The loop follows the ReAct (Reasoning + Acting) pattern, published by Yao et al.

## Stage 4 — Visualizing the ReAct Loop: Thought → Action → Observation → Answer

- The flowchart below shows the complete execution path for a single agent invocation: Each loop back from Observation to Agent is one complete ReAct cycle.
- The LLM decides at every step whether to call another tool or produce a final answer.
- The sequence below zooms into the message-passing protocol between components during one tool call cycle: Two tool calls, three LLM calls, one final answer.
- Every message is appended to the conversation before the next LLM call — this is what "context grows with each step" means in practice.

*Applies **event-driven-architecture** to the decision above.*

```mermaid
flowchart TD
    A([User Query]) --> B[Agent: LLM + Prompt]
    B --> C{Tool call in response?}
    C -- Yes --> D[AgentExecutor: Dispatch Tool]
    D --> E[Tool Function Executes]
    E --> F[Observation appended to history]
    F --> G{Max iterations reached?}
    G -- No --> B
    G -- Yes --> H([Force-stop: return partial answer])
    C -- No --> I([Final Answer returned to user])
```

```mermaid
sequenceDiagram
    participant U as User
    participant AE as AgentExecutor
    participant LLM as LLM (gpt-4o)
    participant T as Tool Function

    U->>AE: invoke({"input": "Weather in Tokyo vs yesterday?"})
    AE->>LLM: [SystemPrompt + UserMessage]
    LLM-->>AE: AIMessage(tool_calls=[get_current_weather])
    AE->>T: get_current_weather(city="tokyo")
    T-->>AE: "Tokyo: 18C, partly cloudy"
    AE->>LLM: ["History + ToolMessage(observation)"]
    LLM-->>AE: AIMessage(tool_calls=[get_yesterday_weather])
    AE->>T: get_yesterday_weather(city="tokyo")
    T-->>AE: "Tokyo yesterday: high of 15C, light rain"
    AE->>LLM: ["History + ToolMessage(observation)"]
    LLM-->>AE: AIMessage(content="Tokyo is 3C warmer today...")
    AE-->>U: {"output": "Tokyo is 3C warmer today..."}
```

**Observable outcome:** Every message is appended to the conversation before the next LLM call — this is what "context grows with each step" means in practice.

## Stage 5 — Real-World Applications of Built-in LangChain Tools

- LangChain ships several ready-to-use tools that cover the most common "reach the internet" use cases.
- Knowing when to use each one saves you from writing boilerplate integrations.
- DuckDuckGo Search is the no-API-key option for web search.
- It returns a handful of snippets and URLs.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langchain_community.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()
print(search.run("LangChain AgentExecutor latest release"))
```

```python
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=1000))
print(wikipedia.run("ReAct prompting technique"))
```

**Observable outcome:** It returns a handful of snippets and URLs.

## Stage 6 — Worked Example: A Research Assistant with Web Search, Calculator, and Stock Price

- This example wires three tools into a single AgentExecutor — a DuckDuckGo search, a custom calculator, and a custom stock price lookup.
- The goal is a "Research Assistant" agent that can answer compound questions.
- Execution trace (abbreviated): The agent correctly sequences three distinct tool types — a mock API, a calculation, and a live web search — and synthesizes them into a coherent answer.
- No user code controlled the sequencing; the LLM decided the order based on its reasoning.

*Applies **low-level-design-systems** to the decision above.*

```python
# NOTE: Install dependencies:
# pip install langchain langchain-openai langchain-community duckduckgo-search

from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent

# --- Tool 1: Web search (built-in, no API key) ---
web_search = DuckDuckGoSearchRun(name="web_search")

# --- Tool 2: Calculator ---
@tool
def calculator(expression: str) -> str:
    """
    Evaluate a mathematical expression and return the numeric result.
    Input must be a valid Python arithmetic expression, e.g. "1500 * 1.07 ** 10".
    Use this for any calculation that requires precise arithmetic.
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})  # restricted eval
        return f"Result: {result}"
    except Exception as exc:
        return f"Error evaluating expression: {exc}"

# --- Tool 3: Stock price lookup ---
@tool
def get_stock_price(ticker: str) -> str:
    """
    Return the current stock price for a given ticker symbol (e.g. AAPL, MSFT, NVDA).
    Use this when the user asks about stock prices or market valuations.
    NOTE: Returns mock data — replace with a real market data API in production.
    """
    mock_prices = {
        "AAPL": 213.07, "TSLA": 172.30, "NVDA": 118.42,
        "MSFT": 415.20, "GOOGL": 175.50, "AMZN": 198.90,
    }
    price = mock_prices.get(ticker.upper())
    if price is None:
        return f"No price data found for ticker: {ticker}"
    return f"{ticker.upper()} current price: ${price:.2f} USD"

# --- Assemble the agent ---
tools = [web_search, calculator, get_stock_price]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
react_prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=8,
    handle_parsing_errors=True,
)

# --- Run a compound research query ---
response = executor.invoke({
    "input": (
        "What is NVIDIA's current stock price? "
        "If I invested $5,000 at that price, how many shares would I own? "
        "Also search for the latest news about NVIDIA's AI chips."
    )
})
print("\n=== FINAL ANSWER ===")
print(response["output"])
```

```
> Entering new AgentExecutor chain...
Thought: I need three pieces of information: NVIDIA stock price, share count calculation, and news.
Action: get_stock_price
Action Input: {"ticker": "NVDA"}
Observation: NVDA current price: $118.42 USD

Thought: Now I'll calculate shares.
Action: calculator
Action Input: {"expression": "5000 / 118.42"}
Observation: Result: 42.22...

Thought: Now I'll search for NVIDIA AI chip news.
Action: web_search
Action Input: NVIDIA AI chips latest news 2026
Observation: [DuckDuckGo snippets about Blackwell GPU architecture, data center demand...]

Thought: I have all three pieces of data. I can compose the final answer.
Final Answer: NVIDIA is currently trading at $118.42. With $5,000 you would own approximately
42.2 shares. Recent news highlights strong demand for Blackwell GPUs in AI data centers...
```

**Observable outcome:** No user code controlled the sequencing; the LLM decided the order based on its reasoning.

## End-to-end outcome

After working through the stages, you should be able to explain:

- Write docstrings for the model, not for humans.
- Every word in a tool docstring is LLM prompt text.
- - LLMs only predict tokens — they cannot fetch data, run calculations, or execute code.
- Tools are the mechanism that bridges language to action.

## Alternatives considered

### AgentExecutor Under the Hood: Runnable Sequences and the Tool-Calling vs. ReAct Distinction

AgentExecutor is built on LangChain's LCEL (LangChain Expression Language) runnable interface.

### Decision Guide: Classic AgentExecutor or LangGraph?

The right tool depends on the complexity of the agent's decision-making surface.


## Failure and degraded modes

### AgentExecutor Trade-offs and Failure Modes

- **Detection:** AgentExecutor is a pragmatic tool with well-understood limitations.
- **User or operator impact:** AgentExecutor is a pragmatic tool with well-understood limitations.



Use prompts, chains, memory, tools, and RAG before adding graph orchestration.

LangChain Basics

## Scenario context

LangChain's `AgentExecutor` is a solid starting point — but it has five hard limits (no branching, no pause/resume, no parallelism, no human-in-the-loop, no crash recovery). LangGraph replaces the implicit loop with an explicit graph, unlocking every one of those capabilities. This post explains the mental model, compares both approaches in code, and helps you decide when to upgrade.

---

You built a LangChain agen…

This walkthrough focuses on **When a Working Agent Hits a Wall**.

Before making a design choice, clarify:

- It queries a knowledge base, calls a few tools, and returns useful answers.
- Then the product requirements arrive.
- "We need it to ask for user approval before it sends the email." You patch it with a flag.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, sorting-and-searching-algorithms, database-indexing

## Stage 1 — Five Places Where AgentExecutor Runs Out of Road

- AgentExecutor implements a ReAct loop: reason, act (pick a tool), observe (read the output), repeat.
- That loop covers a large fraction of real-world agent tasks — but five patterns break it reliably.
- None of these is a bug in LangChain.
- AgentExecutor was designed as a general-purpose, easy-to-use agent loop.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** AgentExecutor was designed as a general-purpose, easy-to-use agent loop.

## Stage 2 — The State Machine Mental Model: From Recipes to Flowcharts

- The key mental shift is this: LangChain chain = a recipe.
- Execute step 1, then step 2, then step 3.
- Perfect for linear transformations.
- LangGraph StateGraph = a flowchart.

*Applies **graph-traversal-algorithms** to the decision above.*

**Observable outcome:** LangGraph StateGraph = a flowchart.

## Stage 3 — The Same Task, Two Ways: Research + Draft an Email

- The LangChain AgentExecutor Version This is clean and gets the job done.
- The LLM will call search web , read the results, then decide to call send email .
- But notice: there is no point in this code where you can intercept the agent before it sends the email.
- Adding an approval step means restructuring the entire calling code.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```python
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain import hub

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@tool
def search_web(query: str) -> str:
    """Search the web for information on a topic."""
    # Replace with a real search integration (Tavily, Serper, etc.)
    return f"[Search results for: {query}] — Key facts retrieved."

@tool
def send_email(recipient: str, subject: str, body: str) -> str:
    """Send an email. USE THIS ONLY AFTER RESEARCH IS COMPLETE."""
    # In reality: sends the email. Here: simulates it.
    return f"Email sent to {recipient} with subject '{subject}'."

tools = [search_web, send_email]
prompt = hub.pull("hwchase17/react")  # standard ReAct prompt

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# This works — but you cannot pause for approval before send_email runs,
# cannot branch to a fallback if search_web returns empty,
# and cannot recover if the process crashes mid-run.
result = executor.invoke({
    "input": "Research climate change impacts and draft an email to the team."
})
print(result["output"])
```

```python
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 1. Typed state schema ────────────────────────────────────────────────────
class EmailAgentState(TypedDict):
    topic: str
    research_results: str
    draft: str
    human_feedback: str
    approved: bool
    retry_count: Annotated[int, operator.add]

# ── 2. Nodes — each is a plain Python function ──────────────────────────────
def research_node(state: EmailAgentState) -> dict:
    """Fetch research on the topic. Falls back gracefully if results are thin."""
    response = llm.invoke([
        HumanMessage(content=f"Summarize key facts about: {state['topic']}")
    ])
    return {"research_results": response.content}

def draft_node(state: EmailAgentState) -> dict:
    """Draft the email body using research results."""
    context = state["research_results"]
    feedback = state.get("human_feedback", "")
    prompt = f"Write a professional team email about {state['topic']}.\nResearch: {context}"
    if feedback:
        prompt += f"\nPrevious feedback to incorporate: {feedback}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"draft": response.content, "approved": False}

def human_approval_node(state: EmailAgentState) -> dict:
    """
    In production: suspend here and wait for a real human API call.
    For this demo: auto-approve after the first retry to avoid an infinite loop.
    """
    if state["retry_count"] > 0:
        # Simulate approval after one retry
        return {"approved": True, "human_feedback": ""}
    # Simulate a rejection on first pass with feedback
    return {
        "approved": False,
        "human_feedback": "Please make the tone more concise.",
        "retry_count": 1,   # operator.add accumulates this
    }

def send_email_node(state: EmailAgentState) -> dict:
    """Send the approved draft — only reachable after human_approval_node approves."""
    print(f"\n✅ Email sent!\n\n{state['draft']}")
    return {}

# ── 3. Routing functions ─────────────────────────────────────────────────────
def route_after_approval(state: EmailAgentState) -> str:
    """Branch on approval flag — either send or loop back to drafting."""
    return "send_email_node" if state["approved"] else "draft_node"

# ── 4. Build the graph ───────────────────────────────────────────────────────
builder = StateGraph(EmailAgentState)

builder.add_node("research_node",       research_node)
builder.add_node("draft_node",          draft_node)
builder.add_node("human_approval_node", human_approval_node)
builder.add_node("send_email_node",     send_email_node)

builder.set_entry_point("research_node")
builder.add_edge("research_node",       "draft_node")
builder.add_edge("draft_node",          "human_approval_node")
builder.add_conditional_edges(
    "human_approval_node",
    route_after_approval,
    {"send_email_node": "send_email_node", "draft_node": "draft_node"},
)
builder.add_edge("send_email_node", END)

# ── 5. Compile with a checkpointer for crash recovery ───────────────────────
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# ── 6. Run it ────────────────────────────────────────────────────────────────
config = {"configurable": {"thread_id": "email-run-001"}}
result = graph.invoke(
    {"topic": "climate change team update", "retry_count": 0,
     "research_results": "", "draft": "", "human_feedback": "", "approved": False},
    config=config,
)
```

**Observable outcome:** Adding an approval step means restructuring the entire calling code.

## Stage 4 — Real-World Applications: Where the Graph Model Earns Its Complexity

- The recipe-vs-flowchart analogy holds in practice.
- Consider three production patterns that LangGraph handles and AgentExecutor cannot: Document approval pipelines — A legal review agent drafts a clause, routes it through compliance review, routes approved clauses to the…
- route back for revision" decision is a conditional edge reading a review status field in state.
- Multi-source research with fallback — A researcher first queries an internal database node.

*Applies **database-indexing** to the decision above.*

**Observable outcome:** Multi-source research with fallback — A researcher first queries an internal database node.

## Stage 5 — LangGraph: The Open-Source Project and Its Architecture

- LangGraph is an open-source Python library from LangChain, Inc., released under the MIT license.
- Read them in order for the full progression, or jump directly to the capability you need.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Read them in order for the full progression, or jump directly to the capability you need.

## End-to-end outcome

It queries a knowledge base, calls a few tools, and returns useful answers.

## Alternatives considered

### Visualizing the Contrast: Linear Loop vs. Branching Graph

The diagrams below show the same "research and draft an email" task implemented in each model.

### LangChain's core abstraction is the **Runnable** — a composable unit with `invoke`, `stream`, and `batch` methods. Chains are built by piping Runnables together: `prompt | llm | parser`. This is expressive and concise for sequential pipelines.

LangGraph's core abstraction is the StateGraph — a directed graph where nodes are Python functions and edges encode the routing logic.

### The Internals: Superstep Execution and State Merging

When you call StateGraph.compile() , LangGraph validates the graph (checks for disconnected nodes, unreachable END , missing entry points) and produces a Pregel-style executor.

### Performance Considerations: When Graph Overhead Is Worth It

LangGraph adds measurable overhead compared to a direct LCEL chain: For the typical agent (state = text + a few flags), serialization adds under a millisecond per superstep.

### Decision Guide: Which Tool for Which Task?

The trigger: the moment your agent needs to decide where to go next based on what it just learned , you have outgrown a pipeline and need a graph.


## Failure and degraded modes

### Trade-offs and Failure Modes: When LangGraph Adds Overhead and When It Doesn't

- **Detection:** A common misconception: "LangGraph replaces LangChain." It does not.
- **User or operator impact:** A common misconception: "LangGraph replaces LangChain." It does not.



## Scenario context

LangGraph adds state, branching, and loops to LLM chains — build stateful agents with graphs, nodes, and typed state.

---

You built a LangChain chain that answers questions. Then you tried to make it ask a follow-up. The chain had no idea what it just said.




This is the **stateless chain problem** — and it trips up almost every developer who moves beyond basic LangChain tutorials. Here is what the failure looks…

This walkthrough focuses on **The Stateless Chain Problem: Why Your Agent Forgets Everything**.

Before making a design choice, clarify:

- You built a LangChain chain that answers questions.
- Then you tried to make it ask a follow-up.
- The chain had no idea what it just said.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, sorting-and-searching-algorithms, event-driven-architecture

## Stage 1 — LangGraph's Mental Model: Graphs, State, Nodes, and Edges

- Before writing any code, you need three clear ideas in your head.
- LangGraph is built from exactly these three concepts — nothing more.
- State is a Python dictionary typed with TypedDict .
- Every node in the graph can read it and write partial updates to it.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Every node in the graph can read it and write partial updates to it.

## Stage 2 — Building Blocks: TypedDict State, Nodes, and How Edges Connect Them

- Let's make everything above concrete with runnable code.
- Here is the minimum viable LangGraph agent — a two-node graph that greets a user and then summarizes the greeting.
- Step 1 — Define your state schema with TypedDict The Annotated[list[str], operator.add] syntax is the first thing that surprises most developers.
- When a node returns {"messages": ["Hello!"]} , LangGraph does not replace the entire messages list — it appends to it.

*Applies **graph-traversal-algorithms** to the decision above.*

```python
from typing import Annotated
from typing_extensions import TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]
    # operator.add means "append new values; never overwrite the whole list"
```

```python
def greet_node(state: AgentState) -> dict:
    """Add a greeting to the message list."""
    return {"messages": ["Hi! How can I help you today?"]}

def summarize_node(state: AgentState) -> dict:
    """Summarize everything that has been said so far."""
    history = " | ".join(state["messages"])
    return {"messages": [f"[Summary]: {history}"]}
```

**Observable outcome:** When a node returns {"messages": ["Hello!"]} , LangGraph does not replace the entire messages list — it appends to it.

## Stage 3 — Deep Dive: How LangGraph Compiles and Executes Your Graph

- The Internals: Pregel-Style Execution Under the Hood When you call StateGraph.compile() , LangGraph validates the graph structure (checks for disconnected nodes, missing entry points, unreachable END ) and produces an e…
- At each superstep, LangGraph identifies the currently active nodes, runs them, collects their partial state updates, and merges those updates into the shared state dict before advancing to the next step.
- The merge strategy for each field is controlled by its annotated reducer — operator.add for append, or overwrite-by-default for unannotated fields.
- LangGraph also serializes the full state after each superstep.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```python
from langgraph.checkpoint.memory import MemorySaver

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

# Each run gets a thread_id — state is persisted per thread
config = {"configurable": {"thread_id": "user-session-42"}}
result = graph.invoke({"messages": []}, config=config)
```

```python
for step in graph.stream({"messages": []}):
    node_name = list(step.keys())[0]
    print(f"✅ Node '{node_name}' completed: {step[node_name]}")
```

**Observable outcome:** LangGraph also serializes the full state after each superstep.

## Stage 4 — Your First Agent: A Research Assistant That Can Search

- Now let's build the worked example.
- The agent takes a user question, decides whether it needs to search the web before answering, runs a search if needed, and synthesizes the final response.
- Here is the graph structure: The router LLM inspects the question and sets a needs search flag.
- If true, the search node runs and injects results into state.

*Applies **event-driven-architecture** to the decision above.*

```mermaid
graph TD
    A([ User Question]) --> B[router_node Decide: search needed?]
    B -->|needs_search = true| C[search_node Fetch results]
    B -->|needs_search = false| D[answer_node Synthesize answer]
    C --> D
    D --> E([END])

    style A fill:#e3f2fd,stroke:#1976D2,color:#000
    style E fill:#e8f5e9,stroke:#388E3C,color:#000
    style B fill:#fff8e1,stroke:#F57F17,color:#000
    style C fill:#f3e5f5,stroke:#7B1FA2,color:#000
    style D fill:#e8f5e9,stroke:#2E7D32,color:#000
```

```python
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, END

# --- 1. State schema ---
class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    search_results: str
    needs_search: bool

# --- 2. LLM (swap freely — see the LLM-agnostic section below) ---
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# --- 3. Node: decide whether to search ---
def router_node(state: ResearchState) -> dict:
    question = state["messages"][-1].content
    prompt = (
        f"Does answering this question require current web information?\n"
        f"Question: {question}\n"
        f"Reply with only: YES or NO"
    )
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"needs_search": "YES" in response.content.upper()}

# --- 4. Node: run the search ---
def search_node(state: ResearchState) -> dict:
    question = state["messages"][-1].content
    # Replace this stub with TavilySearch, Serper, or Bing API in production
    results = f"[Search stub] Top results for: '{question}' — relevant facts retrieved."
    return {"search_results": results}

# --- 5. Node: synthesize the answer ---
def answer_node(state: ResearchState) -> dict:
    context_messages = list(state["messages"])
    if state.get("search_results"):
        context_messages.append(
            HumanMessage(content=f"Use this context to answer: {state['search_results']}")
        )
    context_messages.append(HumanMessage(content="Now answer the original question."))
    response = llm.invoke(context_messages)
    return {"messages": [AIMessage(content=response.content)]}

# --- 6. Routing function ---
def route_search(state: ResearchState) -> str:
    return "search_node" if state.get("needs_search") else "answer_node"

# --- 7. Build and compile the graph ---
builder = StateGraph(ResearchState)

builder.add_node("router_node", router_node)
builder.add_node("search_node", search_node)
builder.add_node("answer_node", answer_node)

builder.set_entry_point("router_node")
builder.add_conditional_edges("router_node", route_search, {
    "search_node": "search_node",
    "answer_node": "answer_node",
})
builder.add_edge("search_node", "answer_node")
builder.add_edge("answer_node", END)

graph = builder.compile()

# --- 8. Run it ---
result = graph.invoke({
    "messages": [HumanMessage(content="What is the current price of Bitcoin?")],
    "search_results": "",
    "needs_search": False,
})
print(result["messages"][-1].content)
```

**Observable outcome:** If true, the search node runs and injects results into state.

## Stage 5 — Real-World Applications: Where Stateful Graphs Show Up in Production

- LangGraph is not a toy framework.
- Stateful graphs power several real production patterns: Customer support routing — A support bot classifies incoming messages: simple FAQs route directly to an answer node; complex issues enter a multi-turn diagnostic l…
- State tracks the conversation history, the issue category, and an escalation flag across every turn without external session storage.
- Code generation with self-repair — An agent generates code, runs a linter node, reads the linter output from state, and conditionally loops back to the LLM to fix errors.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Code generation with self-repair — An agent generates code, runs a linter node, reads the linter output from state, and conditionally loops back to the LLM to fix errors.

## Stage 6 — Practical Example: Making the Graph LLM-Agnostic

- One of LangGraph's cleanest design decisions is that nodes are plain Python functions with no opinion about which LLM you use.
- Swapping providers requires changing exactly one line — the llm assignment at the top.
- This scenario was chosen because it is the most common "day 2" question new LangGraph users ask — and showing it is a one-liner is the best answer.
- As you read through the three provider options, watch how every other line (the research node , the StateGraph , the compile() call) stays identical across all three — that invariance is LangGraph's LLM-agnostic promise…

*Applies **graph-traversal-algorithms** to the decision above.*

```python
# Option 1: OpenAI — best default for production
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Option 2: Anthropic Claude — great for long-context reasoning
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-3-haiku-20240307", temperature=0)

# Option 3: Ollama — fully local, zero API cost, great for development
from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.2", temperature=0)

# The graph, nodes, state schema, and edges are IDENTICAL for all three.
```

```python
def make_router_node(llm):
    """Factory that binds a specific LLM into the router node."""
    def router_node(state: ResearchState) -> dict:
        question = state["messages"][-1].content
        prompt = f"Does answering this require web search?\nQuestion: {question}\nReply YES or NO"
        response = llm.invoke([HumanMessage(content=prompt)])
        return {"needs_search": "YES" in response.content.upper()}
    return router_node

# At graph build time, pick your LLM and inject it:
builder.add_node("router_node", make_router_node(llm))
```

**Observable outcome:** As you read through the three provider options, watch how every other line (the research node , the StateGraph , the compile() call) stays identical across all three — that invariance is LangGraph's LLM-agnostic promise…

## End-to-end outcome

After working through the stages, you should be able to explain:

- Building your first stateful agent surfaces several lessons that are not obvious from the documentation: 1.
- Compile once, invoke many times.

## Alternatives considered

### Decision Guide: StateGraph vs LCEL Chain vs LangChain AgentExecutor

The key signal is decision-making inside the loop .


## Failure and degraded modes

### Trade-offs and Failure Modes: When LangGraph Helps and When It Adds Complexity

- **Detection:** LangGraph solves real problems, but adding it to every project introduces overhead that is not always justified.
- **User or operator impact:** LangGraph solves real problems, but adding it to every project introduces overhead that is not always justified.



## Scenario context

ReAct = Think + Act + Observe, looped as a LangGraph graph — prebuilt or custom.

 B[" Agent Node (LLM call with bind_tools)"]
    B --> C{should_continue}
    C -- "tool_calls present in last AIMessage" --> D[" ToolNode (execute tool functions)"]
    D -- "append ToolMessages to state" --> B
    C -- "no tool_calls (final answer)" --> E([ END])

    style A fill:#4CAF50,color:#fff,stroke:none
    style E fill:#F44336,color:#fff,stroke:none
    style B fill:#2196F3,color:#fff,stroke:#1565C0
    style D fill:#FF9800,color:#fff,stroke:#E65100
    style C fill:#9C27B0,color:#fff,stroke:#6A1B9A
```

**Observable outcome:** What makes this a loop and not a chain is the back-edge from tools → agent .

## Stage 4 — Real-World Applications: Where ReAct Agents Are Deployed Today

- ReAct is not a research curiosity.
- It is the backbone of a wide class of production deployments: Coding assistants with test harnesses (GitHub Copilot Workspace, Devin-style agents): Write code → run linter → fix → run tests → fix → commit.
- The loop terminates when all checks pass or an iteration limit is reached.
- This is exactly the pattern this post's coding assistant example implements.

*Applies **event-driven-architecture** to the decision above.*

**Observable outcome:** This is exactly the pattern this post's coding assistant example implements.

## Stage 5 — Practical Example: Coding Assistant That Iterates Until Tests Pass

- This example implements the opening scenario end-to-end: a coding assistant that writes a Python function, runs the test suite, inspects failures, and iterates until all tests are green — or gives up after 8 attempts.
- The coding-assistant scenario was chosen because it is the clearest possible demonstration of the ReAct loop's core value: the agent cannot solve the task in one shot because it needs to observe real test output before…
- As you read through the execution trace, watch the should continue edge — every iteration where the result is "tools" is one full Think→Act→Observe cycle, and the loop only exits to END when the LLM decides there is not…
- What the loop looks like across 3+ turns: The agent used 5 turns, observed real test output, diagnosed a recursion failure, and applied a targeted fix — none of which is possible in a single shot.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
import os
import subprocess
from typing import Annotated
from typing_extensions import TypedDict

from langchain_core.messages import SystemMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# ── State ──────────────────────────────────────────────────────────────────
class CodingAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int

# ── Tools ──────────────────────────────────────────────────────────────────
@tool
def write_python_file(filename: str, code: str) -> str:
    """Write Python source code to a file. Returns confirmation."""
    with open(filename, "w") as f:
        f.write(code)
    return f"Written {len(code)} chars to '{filename}'."

@tool
def run_pytest(test_file: str) -> str:
    """Run pytest on the given test file. Returns stdout + stderr."""
    result = subprocess.run(
        ["pytest", test_file, "-v", "--tb=short", "--no-header"],
        capture_output=True, text=True, timeout=30
    )
    output = result.stdout + result.stderr
    return output[:4000]  # Guard against extremely verbose output

tools = [write_python_file, run_pytest]

# ── Nodes ──────────────────────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

SYSTEM = SystemMessage(content="""You are an expert Python engineer.
When given a coding task:
1. Write the implementation file using write_python_file.
2. Run the test file using run_pytest.
3. If any tests fail, read the error output carefully, fix the code, and run tests again.
4. Repeat until ALL tests pass, then report "All tests pass." as your final answer.
Do not stop until tests pass or you are truly stuck.""")

MAX_ITERATIONS = 8

def agent_node(state: CodingAgentState):
    if state.get("iterations", 0) >= MAX_ITERATIONS:
        return {
            "messages": [AIMessage(content=f"Stopped after {MAX_ITERATIONS} iterations.")],
            "iterations": state.get("iterations", 0),
        }
    messages = [SYSTEM] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1,
    }

def should_continue(state: CodingAgentState) -> str:
    last = state["messages"][-1]
    if state.get("iterations", 0) >= MAX_ITERATIONS:
        return END
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

tool_node = ToolNode(tools, handle_tool_errors=True)

# ── Graph ──────────────────────────────────────────────────────────────────
workflow = StateGraph(CodingAgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

coding_agent = workflow.compile()

# ── Run ────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    result = coding_agent.invoke({
        "messages": [(
            "user",
            "Implement a fibonacci function in fib.py. "
            "Tests are in test_fib.py. Make all tests pass."
        )],
        "iterations": 0,
    })

    for msg in result["messages"]:
        role = type(msg).__name__
        content = getattr(msg, "content", "") or ""
        if content:
            print(f"[{role}] {content[:200]}")
    print(f"\nCompleted in {result['iterations']} iteration(s).")
```

```
[Turn 1] Agent thinks: "I'll write the fibonacci implementation first."
         Calls: write_python_file("fib.py", "def fib(n):\n    if n <= 1: return n\n    ...")
[Tool]   "Written 89 chars to 'fib.py'."

[Turn 2] Agent thinks: "Now run the tests."
         Calls: run_pytest("test_fib.py")
[Tool]   "FAILED test_fib.py::test_large_input - RecursionError: maximum recursion depth exceeded"

[Turn 3] Agent thinks: "I need an iterative implementation for large inputs."
         Calls: write_python_file("fib.py", "def fib(n):\n    a, b = 0, 1\n    for _ in range(n): a, b = b, a+b\n    return a")
[Tool]   "Written 74 chars to 'fib.py'."

[Turn 4] Agent thinks: "Retry the tests."
         Calls: run_pytest("test_fib.py")
[Tool]   "PASSED: 5 tests in 0.04s"

[Turn 5] Agent produces final answer: "All tests pass. The fibonacci function uses an iterative
         approach and handles inputs up to fib(1000) without recursion errors."
```

**Observable outcome:** What the loop looks like across 3+ turns: The agent used 5 turns, observed real test output, diagnosed a recursion failure, and applied a targeted fix — none of which is possible in a single shot.

## Stage 6 — LangGraph Prebuilt Agents: create_react_agent, create_tool_calling_executor

- LangGraph ships two prebuilt agent constructors.
- Understanding both helps you choose the right level of abstraction.
- create react agent (current standard) create react agent is LLM-agnostic : any model that supports tool calling via .bind tools() works identically.
- This includes ChatOpenAI , ChatAnthropic , ChatOllama , and ChatGroq .

*Applies **graph-traversal-algorithms** to the decision above.*

```python
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic  # swap provider freely

# Works identically with ChatOpenAI, ChatAnthropic, ChatOllama, ChatGroq
agent = create_react_agent(
    model=ChatAnthropic(model="claude-3-5-sonnet-20241022"),
    tools=tools,
    # System prompt injected before messages on every call, not stored in state
    state_modifier=SYSTEM,
)
```

```python
from IPython.display import Image
Image(agent.get_graph().draw_mermaid_png())
```

**Observable outcome:** This includes ChatOpenAI , ChatAnthropic , ChatOllama , and ChatGroq .

## End-to-end outcome

After working through the stages, you should be able to explain:

- The system prompt does not belong in state.
- A common mistake is appending the system prompt as the first message in MessagesState .

## Alternatives considered

### LangGraph gives you two paths to a working ReAct agent. The right choice depends on how much you need to customize the loop.

LangGraph gives you two paths to a working ReAct agent.

### Path 1: The Prebuilt Shortcut with `create_react_agent`

create react agent handles everything: it binds your tools to the LLM, creates the agent node, creates the ToolNode , wires the conditional edge, and compiles the graph.

### Path 2: Building the Loop Manually as a StateGraph

The key difference from create react agent is the should continue function — you own it.

### Decision Guide: create_react_agent vs Custom Loop vs Plan-and-Execute

Decision Guide: create_react_agent vs Custom Loop vs Plan-and-Execute


## Failure and degraded modes

### The Single-Shot Failure: Why One LLM Call Isn't Enough for Complex Tasks

- **Detection:** Your agent is supposed to write a function, run the tests, fix the failures, and repeat until green.
- **User or operator impact:** Your agent is supposed to write a function, run the tests, fix the failures, and repeat until green.

### Performance vs. Cost

- **Detection:** ReAct loops are significantly more expensive than single-shot calls.
- **User or operator impact:** ReAct loops are significantly more expensive than single-shot calls.

### Infinite Loop Failure Mode

- **Detection:** The most dangerous failure mode is an agent that loops without making progress.
- **User or operator impact:** The most dangerous failure mode is an agent that loops without making progress.

### Hallucinated Tool Calls

- **Detection:** The LLM may emit a tool call that references a function that does not exist in your tools list, or passes arguments with the wrong type.
- **User or operator impact:** The LLM may emit a tool call that references a function that does not exist in your tools list, or passes arguments with the wrong type.

### Token Explosion in Long Conversations

- **Detection:** Deep loops on complex tasks can exhaust the context window.
- **User or operator impact:** Deep loops on complex tasks can exhaust the context window.

## Scenario context

Wire @tool, ToolNode, and bind_tools into LangGraph for agents that call APIs at runtime.

Your agent confidently tells you the current stock price of NVIDIA. It's from its training data — six months out of date. The model doesn't know what it doesn't know. It has no concept of "today." Every fact it gives you was frozen at training cutoff, and it will recite that frozen snapshot with the same unwavering confidence…

This walkthrough focuses on **The Stale Knowledge Problem: Why LLMs Need Runtime Tools**.

Before making a design choice, clarify:

- Your agent confidently tells you the current stock price of NVIDIA.
- It's from its training data — six months out of date.
- The model doesn't know what it doesn't know.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, low-level-design-systems, sorting-and-searching-algorithms

## Stage 1 — Tool Fundamentals: @tool, Schemas, and the bind_tools() Pattern

- Before wiring tools into a graph, you need to understand what a "tool" is from LangGraph's perspective: a Python callable decorated with @tool that carries enough metadata for an LLM to decide when and how to call it.
- Defining a Tool with @tool The @tool decorator from langchain core.tools does three things.
- It wraps your function, uses the docstring as the tool description (what the model reads to decide whether to call it), and derives the JSON schema for the function's arguments from Python type hints.
- The docstring is not decoration — it's the model's only window into what this tool does.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langchain_core.tools import tool

@tool
def get_stock_price(ticker: str) -> str:
    """
    Fetch the current stock price for a given ticker symbol.
    Returns a formatted string with the latest price.
    """
    # In production, call a real market data API here
    prices = {"NVDA": "118.42", "AAPL": "213.07", "TSLA": "172.30"}
    price = prices.get(ticker.upper(), "unknown")
    return f"{ticker.upper()} is currently trading at ${price}"
```

```python
from pydantic import BaseModel, Field
from langchain_core.tools import tool

class WebSearchInput(BaseModel):
    query: str = Field(description="The search query string")
    max_results: int = Field(default=5, description="Maximum number of results to return")

@tool("web_search", args_schema=WebSearchInput)
def web_search(query: str, max_results: int = 5) -> str:
    """Search the web for current information on a topic."""
    # In production: call Tavily, SerpAPI, or Brave Search here
    return f"Top {max_results} results for '{query}': [result1, result2, ...]"
```

**Observable outcome:** The docstring is not decoration — it's the model's only window into what this tool does.

## Stage 2 — Wiring Tools into LangGraph: ToolNode, Conditional Routing, and the Tool Loop

- With tools defined and bound to the LLM, the next step is wiring them into the graph.
- This requires three components: a state , an agent node , a ToolNode , and conditional routing between them.
- State and the Agent Node The graph state holds the conversation as a list of messages.
- The agent node calls the LLM with the current state and appends the model's response.

*Applies **graph-traversal-algorithms** to the decision above.*

```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

def agent_node(state: AgentState):
    """Call the LLM with the current message history."""
    response = llm_openai.invoke(state["messages"])
    return {"messages": [response]}
```

```python
tool_node = ToolNode(tools)  # Pass the same list you used in bind_tools()
```

**Observable outcome:** The agent node calls the LLM with the current state and appends the model's response.

## Stage 3 — Deep Dive: How LangGraph Executes Tool Calls

- The Internals When the LLM with bound tools generates a response that includes tool calls, it returns an AIMessage with a tool calls field.
- Each entry in this list is a dict with three keys: id (a unique call identifier), name (the tool function name), and args (a dict of parsed arguments).
- ToolNode iterates this list and dispatches each call to the matching function by name.
- It uses the tool registry it was initialized with — the same tools list passed to ToolNode(tools) .

*Applies **low-level-design-systems** to the decision above.*

```python
# Example AIMessage.tool_calls structure:
[
    {
        "id": "call_abc123",
        "name": "get_stock_price",
        "args": {"ticker": "NVDA"}
    },
    {
        "id": "call_def456",
        "name": "web_search",
        "args": {"query": "NVIDIA Q4 2025 earnings", "max_results": 3}
    }
]
```

**Observable outcome:** It uses the tool registry it was initialized with — the same tools list passed to ToolNode(tools) .

## Stage 4 — The Tool Calling Loop: Graph Diagram

- The core pattern is a tight loop between an agent node and a tool executor, gated by a conditional router: The agent loop: the LLM reasons → calls tools if needed → receives results → reasons again.
- The cycle exits when the model produces a direct answer.
- Every iteration of the loop passes the full message history to the LLM, so the model always has context on what tools it called and what results they returned.
- This is how multi-step reasoning works: the agent sees its own prior actions and builds on them.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```mermaid
flowchart TD
    A([User Input]) --> B[Agent Node LLM + bind_tools]
    B --> C{tools_condition}
    C -- has tool_calls --> D[ToolNode Execute tools]
    C -- no tool_calls --> E([Final Response])
    D --> B
    style A fill:#e8f4f8,stroke:#2196f3
    style E fill:#e8f5e9,stroke:#4caf50
    style C fill:#fff3e0,stroke:#ff9800
    style D fill:#fce4ec,stroke:#e91e63
```

**Observable outcome:** This is how multi-step reasoning works: the agent sees its own prior actions and builds on them.

## Stage 5 — Real-World Applications: How Production Agents Use Tool Calling

- Financial Research Assistants A hedge fund assistant needs to answer: "Should we increase our NVIDIA position given today's macro environment?" A bare LLM would hallucinate a confident answer based on stale training dat…
- With tools, the agent calls get stock price , get earnings data , and web search in parallel, receives live data, then synthesizes a grounded analysis.
- Input: "Evaluate NVIDIA given today's market." Process: Parallel tool calls → stock price + recent news + analyst ratings fetched live Output: Structured memo citing today's actual data, not training-cutoff prices.
- Customer Support Bots with CRM Access A support agent needs to check an order's shipping status.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Customer Support Bots with CRM Access A support agent needs to check an order's shipping status.

## Stage 6 — Practical Example: Market Research Agent with Parallel Tool Calls

- This example demonstrates the full tool-calling loop: tool definition, LLM binding, ToolNode wiring, and parallel execution in a single runnable agent.
- The market research scenario was chosen because it requires live data from multiple independent sources simultaneously — exactly the case where LangGraph's parallel ToolNode execution pays off over a sequential loop.
- As you read through the code, watch for step 2 in the execution trace: the LLM emits two tool calls in a single AIMessage , which ToolNode executes concurrently — that is the parallel tool calling pattern in action.
- Here is the complete market research agent with three tools — web search , get stock price , and summarize findings — with parallel execution and graceful error handling.

*Applies **graph-traversal-algorithms** to the decision above.*

```python
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# ── Tool definitions ────────────────────────────────────────────────
@tool
def web_search(query: str) -> str:
    """Search the web for recent news and analysis on a topic."""
    # Replace with Tavily or SerpAPI in production
    return f"[Search results for '{query}']: Market analysts expect strong Q1. Supply constraints easing."

@tool
def get_stock_price(ticker: str) -> str:
    """Get the current stock price for a publicly traded company by ticker symbol."""
    mock_prices = {"NVDA": "118.42", "AAPL": "213.07", "MSFT": "415.30"}
    price = mock_prices.get(ticker.upper(), "Price unavailable")
    return f"{ticker.upper()}: ${price}"

@tool
def summarize_findings(raw_data: str) -> str:
    """Condense multiple data points into a brief investment summary."""
    return f"Summary: Based on provided data — {raw_data[:80]}... — outlook is cautiously positive."

tools = [web_search, get_stock_price, summarize_findings]

# ── LLM with bound tools ─────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

# ── State ─────────────────────────────────────────────────────────────
class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]

# ── Agent node ────────────────────────────────────────────────────────
def agent_node(state: ResearchState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ── Graph assembly ────────────────────────────────────────────────────
tool_node = ToolNode(tools)

builder = StateGraph(ResearchState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()

# ── Run ───────────────────────────────────────────────────────────────
result = graph.invoke(
    {"messages": [HumanMessage(
        content="Research NVIDIA (NVDA): get the current price and recent news. Then summarize."
    )]},
    config={"recursion_limit": 10}
)

for msg in result["messages"]:
    print(f"[{msg.__class__.__name__}] {msg.content[:120]}")
```

**Observable outcome:** Here is the complete market research agent with three tools — web search , get stock price , and summarize findings — with parallel execution and graceful error handling.

## End-to-end outcome

After working through the stages, you should be able to explain:

- Docstrings are the model's tool selection logic.
- A vague docstring like "Get price" will lead the model to call the wrong tool or skip it entirely.

## Alternatives considered

### Decision Guide: ToolNode vs Custom Tool Execution vs LangChain AgentExecutor

Decision Guide: ToolNode vs Custom Tool Execution vs LangChain AgentExecutor


## Failure and degraded modes

### Hallucinated Tool Calls

- **Detection:** LLMs occasionally invent tool call arguments that violate the schema, or call a tool with plausible-sounding but wrong input (e.g., passing "NVIDIA" instead of "NVDA" to a ticker lookup).
- **User or operator impact:** LLMs occasionally invent tool call arguments that violate the schema, or call a tool with plausible-sounding but wrong input (e.g., passing "NVIDIA" instead of "NVDA" to a ticker lookup).

### API Failures and Transient Errors

- **Detection:** If get stock price raises an exception, the default ToolNode wraps it in an error message and feeds it back to the agent.
- **User or operator impact:** If get stock price raises an exception, the default ToolNode wraps it in an error message and feeds it back to the agent.

### Infinite Tool Loops

- **Detection:** An agent can get stuck in a loop: call tool → get result → call tool again in the next step indefinitely.
- **User or operator impact:** An agent can get stuck in a loop: call tool → get result → call tool again in the next step indefinitely.

### Latency Creep at Scale

- **Detection:** Every tool call adds a network round trip.
- **User or operator impact:** Every tool call adds a network round trip.



## Scenario context

Pause LangGraph agents mid-run with interrupt(), get human approval, resume with Command.

---

Your autonomous coding agent refactored the authentication module while you were in a meeting. It looked right to the LLM. It broke production.




This is not a hypothetical. As LangGraph agents gain access to real tools — GitHub APIs, database write operations, email senders, cloud billing — each step they take can have…

This walkthrough focuses on **The Autonomous Agent Risk: When Acting Without Permission Goes Wrong**.

Before making a design choice, clarify:

- Your autonomous coding agent refactored the authentication module while you were in a meeting.
- As LangGraph agents gain access to real tools — GitHub APIs, database write operations, email senders, cloud billing — each step they take can have irreversible real-world consequences .
- An agent that deletes the wrong S3 bucket, commits a breaking change to main, or triggers a $2,000 API bill does not stop to ask for permission.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, jvm-internals-and-garbage-collection, stack-and-queue-data-structures

## Stage 1 — HITL Fundamentals: interrupt(), NodeInterrupt, and the Checkpointer Requirement

- Three primitives make human-in-the-loop possible in LangGraph.
- Understanding what each one does — and what it requires — prevents the most common configuration mistakes.
- interrupt() — The Primary Pause Mechanism interrupt() is a function you call inside a node to pause the graph and return control to the caller with a payload.
- The graph stops at exactly that line.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langgraph.types import interrupt

def approval_node(state: AgentState):
    # Graph pauses here; payload is sent to the caller
    human_decision = interrupt({
        "question": "Approve this action?",
        "proposed_action": state["proposed_action"],
    })
    # Execution resumes here after the human responds
    return {"approved": human_decision == "approve"}
```

```python
from langgraph.errors import NodeInterrupt

def review_node(state: AgentState):
    if not state.get("human_reviewed"):
        # Older pattern: raise an exception to interrupt
        raise NodeInterrupt(f"Please review proposed changes: {state['proposed_action']}")
    return {"status": "reviewed"}
```

**Observable outcome:** The graph stops at exactly that line.

## Stage 2 — Building an Approval Workflow: Pause, Inspect, Approve, Resume

- The full HITL lifecycle has four phases.
- Here is the complete flow in code.
- The checkpointer (D → J) is the invisible bridge that makes both invoke() calls part of the same workflow.
- Phase 1 — Invoke the graph (first run): The graph runs until it hits interrupt() .

*Applies **graph-traversal-algorithms** to the decision above.*

```mermaid
graph TD
    A([Start: invoke graph]) --> B[analyze_node runs]
    B --> C[approval_node hits interrupt]
    C --> D{Checkpointer saves frozen state}
    D --> E([Return __interrupt__ to caller])
    E --> F[Human reviews payload]
    F --> G{Decision}
    G -- approve --> H[invoke Command resume=approve]
    G -- reject --> I[invoke Command resume=reject]
    H --> J[Checkpointer restores state]
    I --> J
    J --> K[approval_node resumes; interrupt returns decision]
    K --> L[apply_node executes]
    L --> M([Graph completes])
```

```python
config = {"configurable": {"thread_id": "deploy-approval-001"}}

result = graph.invoke({"task": "deploy service to production"}, config)
```

**Observable outcome:** Phase 1 — Invoke the graph (first run): The graph runs until it hits interrupt() .

## Stage 3 — Deep Dive: How LangGraph Freezes and Resumes Graph Execution

- The Internals When interrupt() is called inside a node, three things happen in sequence: 1.
- interrupt() raises a special internal exception ( GraphInterrupt ) that the LangGraph executor catches at the top of the execution loop.
- This unwinds the call stack cleanly without corrupting state.
- Before returning to the caller, the executor serializes the full state snapshot — including the exact values of all state keys at the moment of interruption — and writes it to the checkpointer under the current thread i…

*Applies **jvm-internals-and-garbage-collection** to the decision above.*

```python
def deploy_node(state: AgentState):
    # ✅ Safe: pure computation before interrupt
    proposed = build_deploy_plan(state["task"])

    # Graph pauses here — code above runs again on resume (idempotent is fine)
    decision = interrupt({"plan": proposed})

    # ✅ Safe: side effects AFTER interrupt, only execute once (after resume)
    if decision == "approve":
        call_deploy_api(proposed)  # Only called post-resume
    return {"deployed": decision == "approve"}
```

**Observable outcome:** Before returning to the caller, the executor serializes the full state snapshot — including the exact values of all state keys at the moment of interruption — and writes it to the checkpointer under the current thread i…

## Stage 4 — The Interrupt-Resume Cycle: Full Lifecycle in One Diagram

- The diagram makes two non-obvious points explicit: the checkpointer is the bridge between the two invoke() calls, and approval node is re-entered from its beginning on resume — it does not continue from a saved instruct…

*Applies **stack-and-queue-data-structures** to the decision above.*

```mermaid
sequenceDiagram
    participant App as Application Layer
    participant Graph as LangGraph Executor
    participant Node as Agent Node
    participant CP as Checkpointer
    participant Human as Human Reviewer

    App->>Graph: invoke(initial_state, config)
    Graph->>Node: execute analyze_node()
    Node-->>Graph: returns state update
    Graph->>Node: execute approval_node()
    Node->>Graph: interrupt(proposed_action)
    Graph->>CP: save_checkpoint(thread_id, state_snapshot)
    Graph-->>App: return {"__interrupt__": [payload]}

    App->>Human: display(payload[proposed_action])
    Human-->>App: "approve" / "reject" / edited_value

    App->>Graph: invoke(Command(resume=decision), config)
    Graph->>CP: restore_checkpoint(thread_id)
    Graph->>Node: re-enter approval_node(); interrupt() returns decision
    Node-->>Graph: returns {"approved": True}
    Graph->>Node: execute apply_node()
    Node-->>Graph: returns {"applied": True}
    Graph-->>App: final state
```

**Observable outcome:** The diagram makes two non-obvious points explicit: the checkpointer is the bridge between the two invoke() calls, and approval node is re-entered from its beginning on resume — it does not continue from a saved instruct…

## Stage 5 — Real-World Applications: Where Human-in-the-Loop Is Non-Negotiable

- Autonomous DevOps Pipelines A LangGraph agent monitors production metrics, detects anomalies, and generates a runbook of remediation steps.
- Steps like "restart service" might auto-execute, but "scale down the database replica" triggers an interrupt() .
- The on-call engineer receives the proposed change in Slack, approves or rejects it with one click, and the agent resumes.
- Input: Prometheus alert payload, current service topology Interrupt payload: {"action": "scale down", "target": "db-replica-3", "reason": "p99 latency normal"} Human decision: Approve or reject with a comment added via…

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Input: Prometheus alert payload, current service topology Interrupt payload: {"action": "scale down", "target": "db-replica-3", "reason": "p99 latency normal"} Human decision: Approve or reject with a comment added via…

## Stage 6 — Practical Example: PR Review Agent That Asks Before Applying Changes

- This agent demonstrates the complete HITL lifecycle in a scenario where the stakes are high enough to justify every pause: code diffs that, if applied incorrectly, break production.
- The PR review scenario was chosen because it maps cleanly to the three primitives covered in this post — interrupt() for pausing, Command(resume=...) for resuming, and update state() for human correction — each one trig…
- As you read through the nodes, watch for the interrupt() call inside review node and trace how the payload flows back as the return value when the graph resumes — that handoff is the mechanism that makes the entire patt…
- Running the agent — Phase 1 (invoke and receive interrupt): Phase 2 (human reviews and resumes): The thread ID "pr-42-review" binds the two invoke() calls together.

*Applies **graph-traversal-algorithms** to the decision above.*

```python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

# --- State schema ---
class PRReviewState(TypedDict):
    pr_number: int
    diff: str
    analysis: str
    proposed_changes: list[str]
    approved: bool
    applied: bool

llm = ChatOpenAI(model="gpt-4o")

# --- Node 1: Analyze the diff ---
def analyze_pr(state: PRReviewState) -> dict:
    response = llm.invoke(
        f"Analyze this code diff and identify specific issues:\n{state['diff']}"
    )
    return {"analysis": response.content}

# --- Node 2: Generate concrete change proposals ---
def propose_changes(state: PRReviewState) -> dict:
    response = llm.invoke(
        f"Based on this analysis, write 3 specific, actionable code changes:\n{state['analysis']}"
    )
    changes = [line for line in response.content.split("\n") if line.strip()]
    return {"proposed_changes": changes[:3]}

# --- Node 3: Interrupt for human approval ---
def human_approval(state: PRReviewState) -> dict:
    # Graph pauses here — payload goes to the caller
    decision = interrupt({
        "message": "Review proposed changes. Reply 'approve' to proceed or 'reject' to cancel.",
        "pr_number": state["pr_number"],
        "proposed_changes": state["proposed_changes"],
    })
    return {"approved": decision.strip().lower() == "approve"}

# --- Node 4: Apply approved changes ---
def apply_changes(state: PRReviewState) -> dict:
    if not state["approved"]:
        print(f"PR #{state['pr_number']}: changes rejected by reviewer.")
        return {"applied": False}

    # Re-validate before committing (guard against stale state)
    print(f"Applying {len(state['proposed_changes'])} changes to PR #{state['pr_number']}:")
    for change in state["proposed_changes"]:
        print(f"  → {change}")
    # In production: call GitHub API, run git apply, post review comment, etc.
    return {"applied": True}

# --- Build the graph ---
builder = StateGraph(PRReviewState)
builder.add_node("analyze_pr", analyze_pr)
builder.add_node("propose_changes", propose_changes)
builder.add_node("human_approval", human_approval)
builder.add_node("apply_changes", apply_changes)

builder.add_edge(START, "analyze_pr")
builder.add_edge("analyze_pr", "propose_changes")
builder.add_edge("propose_changes", "human_approval")
builder.add_edge("human_approval", "apply_changes")
builder.add_edge("apply_changes", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
```

```python
config = {"configurable": {"thread_id": "pr-42-review"}}
sample_diff = "- def login(username, password):\n+ def login(username: str, password: str) -> bool:"

result = graph.invoke(
    {"pr_number": 42, "diff": sample_diff, "approved": False, "applied": False},
    config
)

# The graph paused at human_approval
payload = result["__interrupt__"][0].value
print(payload["message"])           # "Review proposed changes..."
print(payload["proposed_changes"])  # ["Add type annotations", ...]
```

**Observable outcome:** Running the agent — Phase 1 (invoke and receive interrupt): Phase 2 (human reviews and resumes): The thread ID "pr-42-review" binds the two invoke() calls together.

## End-to-end outcome

After working through the stages, you should be able to explain:

- Always persist your checkpointer before going to production.
- A process restart during an interrupt wipes all frozen thread states.

## Alternatives considered

### Decision Guide: Full Autonomy vs Interrupt-on-Action vs Step-by-Step Approval

Decision Guide: Full Autonomy vs Interrupt-on-Action vs Step-by-Step Approval


## Failure and degraded modes

### Performance vs. Safety

- **Detection:** Adding an interrupt to a workflow introduces unbounded latency .
- **User or operator impact:** Adding an interrupt to a workflow introduces unbounded latency .

### Failure Mode 1 — The Deadlocked Graph

- **Detection:** A thread is interrupted and nobody resumes it.
- **User or operator impact:** A thread is interrupted and nobody resumes it.

### Failure Mode 2 — Stale State on Late Resume

- **Detection:** The LLM proposed changes based on a codebase snapshot from 09:00.
- **User or operator impact:** The LLM proposed changes based on a codebase snapshot from 09:00.

### Failure Mode 3 — Multiple Interrupts in One Thread

- **Detection:** If a graph has two interrupt() calls in sequence, each one halts and resumes independently, in order.
- **User or operator impact:** If a graph has two interrupt() calls in sequence, each one halts and resumes independently, in order.



Model stateful, durable agent loops with tool calls and human checkpoints.

LangGraph Workflows

## Scenario context

Checkpointers + thread IDs give LangGraph agents persistent memory across turns and sessions.
---

Your customer support agent is on its third message with a user. The user says: *"As I mentioned before, my order number is 9847."* The agent replies: *"Could you please provide your order number so I can look that up?"*




The agent hasn't forgotten. It never knew. Every call to `graph.invoke()` without a checkpointe…

This walkthrough focuses on **The Amnesia Problem: Why Stateless Agents Frustrate Users**.

Before making a design choice, clarify:

- Your customer support agent is on its third message with a user.
- The user says: "As I mentioned before, my order number is 9847." The agent replies: "Could you please provide your order number so I can look that up?" The agent hasn't forgotten.
- Every call to graph.invoke() without a checkpointer starts with a completely empty state — zero messages, zero context, no trace of anything said before.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, api-gateway, jvm-internals-and-garbage-collection

## Stage 1 — Checkpointers: InMemorySaver, SqliteSaver, and PostgresSaver

- A checkpointer is an object you attach at compile() time.
- After every node runs, LangGraph calls the checkpointer to write the current state to storage.
- On the next invoke() with the same thread ID, LangGraph calls the checkpointer to read the latest snapshot before the graph starts running.
- LangGraph ships three checkpointers out of the box: InMemorySaver — Development and Unit Tests State lives in a Python dictionary in RAM.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
```

```python
from langgraph.checkpoint.sqlite import SqliteSaver

# Context-manager handles connection lifecycle
with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke({"messages": [...]}, config=config)
```

**Observable outcome:** LangGraph ships three checkpointers out of the box: InMemorySaver — Development and Unit Tests State lives in a Python dictionary in RAM.

## Stage 2 — Deep Dive: How LangGraph Checkpointing Works Under the Hood

- The Internals When LangGraph compiles a graph with a checkpointer, it wraps every node execution in a checkpoint lifecycle: 1.
- Before the graph runs: LangGraph calls checkpointer.get tuple(config) .
- If a checkpoint exists for the given thread id , it deserializes the state and uses it as the starting values.
- If not, the graph starts fresh.

*Applies **graph-traversal-algorithms** to the decision above.*

```python
# The config dict is the entire addressing scheme for LangGraph persistence
config = {
    "configurable": {
        "thread_id": "user-alice-session-42",   # conversation namespace
        "checkpoint_id": None,                   # None = latest; set a UUID to time-travel
    }
}
```

**Observable outcome:** If not, the graph starts fresh.

## Stage 3 — Thread Isolation and State Flow

- The diagram below shows two users — Alice and Bob — both interacting with the same compiled graph.
- Their state never touches because thread id keeps every checkpoint in its own namespace within the same checkpointer backend.
- Two conversations share one graph instance and one checkpointer backend, but their state is completely isolated by thread id .
- Each arrow to the checkpointer is a separate row in the checkpoints table.

*Applies **api-gateway** to the decision above.*

```mermaid
graph TD
    subgraph Alice[Thread: user-alice]
        A1([invoke turn 1]) --> A2[classify_intent node]
        A2 --> A3[fetch_order node]
        A3 --> A4[respond node]
        A4 --> A5([invoke turn 2 resumes from A4 snapshot])
    end

    subgraph Bob[Thread: user-bob]
        B1([invoke turn 1]) --> B2[classify_intent node]
        B2 --> B3[respond node]
        B3 --> B4([invoke turn 2 resumes from B3 snapshot])
    end

    CP[(Checkpointer PostgresSaver)]

    A4 -- "writes checkpoint thread=user-alice" --> CP
    B3 -- "writes checkpoint thread=user-bob" --> CP
    A5 -- "reads checkpoint thread=user-alice" --> CP
    B4 -- "reads checkpoint thread=user-bob" --> CP
```

```python
# Snapshot of Alice's current conversation state
state = graph.get_state({"configurable": {"thread_id": "user-alice"}})
print(state.values["messages"])   # full message list
print(state.next)                 # next node(s) that would run

# Full audit trail — every checkpoint in reverse-chronological order
history = list(graph.get_state_history({"configurable": {"thread_id": "user-alice"}}))
for snapshot in history:
    print(f"Step {snapshot.config['configurable']['checkpoint_id']}: "
          f"next={snapshot.next}, messages={len(snapshot.values['messages'])}")
```

**Observable outcome:** Each arrow to the checkpointer is a separate row in the checkpoints table.

## Stage 4 — Real-World Applications: How Production Agents Use Persistent Memory

- Case Study 1 — E-Commerce Support Agent An online retailer runs a LangGraph support agent with PostgresSaver .
- When a customer opens a chat: - Input: thread id = f"customer-{customer id}-{ticket id}" - Process: The graph loads prior messages for this ticket.
- If the customer said "I already tried resetting my password" three turns ago, the agent routes to escalation rather than suggesting the same fix again.
- - Output: First-contact resolution rate improved ~18% in one deployment because agents stopped repeating questions.

*Applies **jvm-internals-and-garbage-collection** to the decision above.*

```python
# Agent drafts report, hits "awaiting_approval" interrupt node, and pauses
result = graph.invoke({"messages": [...]}, config=config)
# result.next == ("awaiting_approval",) — graph is suspended

# Hours later, manager approves via dashboard — graph resumes from exact checkpoint
graph.invoke(
    Command(resume={"approved": True, "approver": "manager@co.com"}),
    config=config
)
```

**Observable outcome:** - Output: First-contact resolution rate improved ~18% in one deployment because agents stopped repeating questions.

## Stage 5 — LangGraph Memory Store: Long-Term Semantic Memory Beyond the Conversation

- The BaseStore interface is LangGraph's answer to memory that outlives any single thread.
- It is a key-value store organized into namespaces (tuples of strings) with keys (string identifiers) and values (arbitrary dicts).
- For production, replace InMemoryStore with AsyncPostgresStore : Message Trimming to Prevent Context Overflow As conversations grow, trim old messages before passing them to the LLM.
- LangChain's trim messages utility handles this cleanly: For very long-running agents, combine trimming with a summarization node that periodically condenses old messages into a single summary AIMessage , reducing histor…

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# Write a user preference (namespace, key, value)
store.put(("user_profiles", "alice"), "preferences", {
    "language": "en",
    "contact_channel": "email",
    "last_order": "9847"
})

# Read it back in any future session
results = store.search(("user_profiles", "alice"))
if results:
    prefs = results[0].value
    print(prefs["contact_channel"])   # "email"

# Update — put() is upsert, same namespace+key overwrites
store.put(("user_profiles", "alice"), "preferences", {
    **prefs,
    "last_order": "10234"
})
```

```python
from langgraph.store.postgres.aio import AsyncPostgresStore

store = AsyncPostgresStore.from_conn_string(DB_URI)
await store.setup()   # creates the store table on first run
```

**Observable outcome:** LangChain's trim messages utility handles this cleanly: For very long-running agents, combine trimming with a summarization node that periodically condenses old messages into a single summary AIMessage , reducing histor…

## End-to-end outcome

After working through the stages, you should be able to explain:

- Pick your checkpointer before you write any agent code.
- Switching checkpointers later means migrating stored state — InMemorySaver checkpoints cannot be imported into SqliteSaver .

## Alternatives considered

### Memory Fundamentals: Short-Term State vs Long-Term Store vs External Memory

LangGraph offers three distinct memory layers, and conflating them is the most common source of confusion when building persistent agents.


## Failure and degraded modes

### Trade-offs and Failure Modes: Storage Costs, Context Overflow, and Stale Memory

- **Detection:** Storage Costs Compound Quickly What happens on Turn 2: LangGraph loads the checkpoint from Turn 1 (which already has Alice's first message and the assistant's reply).
- **User or operator impact:** Storage Costs Compound Quickly What happens on Turn 2: LangGraph loads the checkpoint from Turn 1 (which already has Alice's first message and the assistant's reply).



## Scenario context

Split work across specialist agents — supervisor routing beats one overloaded generalist every time.

---

Your research agent is writing a 20-page report. It has 15 tools. Its context window is full by page 3. The last 17 pages are hallucinated.




This is not a model quality problem. It is a structural problem: one agent trying to do everything hits three hard limits.

**1. The context window ceiling.**
Every too…

This walkthrough focuses on **The Context Ceiling: Why One Agent Can't Do Everything**.

Before making a design choice, clarify:

- Your research agent is writing a 20-page report.
- Its context window is full by page 3.
- The last 17 pages are hallucinated.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, llm-prompt-engineering, sorting-and-searching-algorithms

## Stage 1 — Deep Dive: State Isolation, Subgraphs, and the Send API

- The Internals How subgraph state boundaries work.
- When you wrap a specialist agent as a subgraph instead of a plain node, it gets its own private TypedDict .
- The parent graph and the subgraph communicate only at explicit boundary points — the entry edge (parent state → subgraph input mapping) and the exit edge (subgraph output → parent state mapping).
- This means a subgraph can have its own tool calls , scratchpad , or intermediate steps fields that never pollute the parent graph's state.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from langgraph.graph import StateGraph

class ResearcherState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    search_queries: list[str]       # private to researcher
    sources_found: list[str]        # private to researcher

# Build the researcher as a standalone graph
researcher_builder = StateGraph(ResearcherState)
# ... add nodes, tools, loops ...
researcher_subgraph = researcher_builder.compile()

# Mount the subgraph as a node in the parent
def researcher_node(state: ResearchState) -> dict:
    # Map parent state → subgraph input
    sub_result = researcher_subgraph.invoke({
        "messages": state["messages"],
        "search_queries": [],
        "sources_found": []
    })
    # Map subgraph output → parent state
    return {"messages": sub_result["messages"]}
```

```mermaid
sequenceDiagram
    participant S as Supervisor
    participant LLM as Router LLM
    participant R as Researcher
    participant FC as Fact Checker
    participant W as Writer

    S->>LLM: Current state + routing prompt
    LLM-->>S: Command(goto="researcher")
    S->>R: invoke(state)
    R-->>S: updated messages (sources found)
    S->>LLM: Updated state + routing prompt
    LLM-->>S: Command(goto="fact_checker")
    S->>FC: invoke(state)
    FC-->>S: updated messages (verified claims)
    S->>LLM: Updated state + routing prompt
    LLM-->>S: Command(goto="writer")
    S->>W: invoke(state)
    W-->>S: final_report
    S->>LLM: Updated state + routing prompt
    LLM-->>S: Command(goto=END)
```

**Observable outcome:** This means a subgraph can have its own tool calls , scratchpad , or intermediate steps fields that never pollute the parent graph's state.

## Stage 2 — Scaling Multi-Agent Systems: Edge Cases, Optimizations, and What Teams Get Wrong

- Scaling to 10+ Agents Adding more agents is not free.
- Every new specialist adds a routing option that the supervisor LLM must evaluate.
- Beyond 7–10 agents, the routing prompt grows long enough to degrade routing quality — the model starts making errors about which agent is appropriate.
- Mitigations: - Hierarchical supervisors : group agents into sub-teams, each managed by a sub-supervisor.

*Applies **graph-traversal-algorithms** to the decision above.*

**Observable outcome:** Mitigations: - Hierarchical supervisors : group agents into sub-teams, each managed by a sub-supervisor.

## Stage 3 — The Deep Research System Architecture

- The following diagram shows the full deep research system: the supervisor fans out to the researcher and fact-checker in parallel, then sequentially routes to the writer, and finally exits.
- A financial services firm built a multi-agent pipeline for regulatory document analysis.
- The system decomposes each filing into four parallel workstreams: a table extraction agent , a risk clause agent , a cross-reference agent , and a summarization agent .
- Each agent runs on a focused 20k-token context window rather than feeding the 400-page document to one model.

*Applies **llm-prompt-engineering** to the decision above.*

**Observable outcome:** Each agent runs on a focused 20k-token context window rather than feeding the 400-page document to one model.

## Stage 4 — Practical Example: Deep Research System with Three Specialist Agents

- This example demonstrates a deep research pipeline where specialist agents run in parallel via Send -based fan-out — exactly the pattern the post's 🧠 Internals section describes as the supervisor's coordination mechani…
- The deep-research scenario was chosen because it has three fundamentally different agent roles (researcher, fact-checker, writer) that map clearly to the three supervisor routing decisions: fan-out to two agents, wait f…
- As you read through the supervisor function, watch the Send calls in the fan out to specialists routing — each Send("researcher", ...) and Send("fact checker", ...) is a parallel task dispatch, and results complete fire…
- What this demonstrates: - The supervisor inspects explicit state flags ( researcher done , fact checker done ) rather than relying on the LLM to infer completion — this makes routing deterministic and auditable.

*Applies **sorting-and-searching-algorithms** to the decision above.*

```python
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.types import Command, Send
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage, SystemMessage
from langchain_core.tools import tool

# ── State ──────────────────────────────────────────────────────────────────────

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    researcher_done: bool
    fact_checker_done: bool
    final_report: str

# ── Tools ─────────────────────────────────────────────────────────────────────

@tool
def web_search(query: str) -> str:
    """Search the web for up-to-date information on a topic."""
    return f"[Web results for '{query}': Found 5 relevant sources about recent developments.]"

@tool
def fact_check(claim: str) -> str:
    """Verify whether a specific claim is supported by cited sources."""
    return f"[Fact check for '{claim}': Supported by 3 sources, 1 contested.]"

# ── Agent LLMs ────────────────────────────────────────────────────────────────

researcher_llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools([web_search])
fact_checker_llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools([fact_check])
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.4)
router_llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── Worker Nodes ──────────────────────────────────────────────────────────────

def researcher_node(state: ResearchState) -> dict:
    sys_prompt = SystemMessage(content=(
        "You are a research specialist. Use web_search to gather current, "
        "accurate information. Be thorough and cite your sources."
    ))
    response = researcher_llm.invoke([sys_prompt] + state["messages"])
    return {
        "messages": [response],
        "researcher_done": True
    }

def fact_checker_node(state: ResearchState) -> dict:
    sys_prompt = SystemMessage(content=(
        "You are a fact-checking specialist. Review all claims in the conversation "
        "and use fact_check to verify each one. Flag unverified claims explicitly."
    ))
    response = fact_checker_llm.invoke([sys_prompt] + state["messages"])
    return {
        "messages": [response],
        "fact_checker_done": True
    }

def writer_node(state: ResearchState) -> dict:
    sys_prompt = SystemMessage(content=(
        "You are a professional report writer. Using the research findings and "
        "fact-check results in this conversation, write a comprehensive, "
        "well-structured final report. Be clear, concise, and accurate."
    ))
    response = writer_llm.invoke([sys_prompt] + state["messages"])
    return {
        "messages": [response],
        "final_report": response.content
    }

# ── Supervisor Node ────────────────────────────────────────────────────────────

ROUTING_PROMPT = """You are a research supervisor. Workers available:
- researcher: gathers information (run first)
- fact_checker: verifies claims (run after researcher)
- writer: writes the final report (run after fact_checker)
- FINISH: end the workflow (run after writer has produced a report)

Current state:
- researcher_done: {researcher_done}
- fact_checker_done: {fact_checker_done}
- has_final_report: {has_final_report}

What is the next step? Reply with exactly one word.
"""

def supervisor_node(state: ResearchState) -> Command[
    Literal["researcher", "fact_checker", "writer", "__end__"]
]:
    prompt = ROUTING_PROMPT.format(
        researcher_done=state.get("researcher_done", False),
        fact_checker_done=state.get("fact_checker_done", False),
        has_final_report=bool(state.get("final_report")),
    )
    response = router_llm.invoke(
        [SystemMessage(content=prompt)] + state["messages"]
    )
    decision = response.content.strip().lower()

    if decision == "finish":
        return Command(goto=END)

    valid = {"researcher", "fact_checker", "writer"}
    if decision not in valid:
        # Fallback: if LLM returns unexpected output, force-finish
        return Command(goto=END)

    return Command(goto=decision, update={"next_agent": decision})

# ── Graph Assembly ─────────────────────────────────────────────────────────────

builder = StateGraph(ResearchState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("fact_checker", fact_checker_node)
builder.add_node("writer", writer_node)

# Workers return to supervisor after completing
builder.add_edge("researcher", "supervisor")
builder.add_edge("fact_checker", "supervisor")
builder.add_edge("writer", "supervisor")

builder.set_entry_point("supervisor")

graph = builder.compile()

# ── Run ────────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    initial_state = {
        "messages": [HumanMessage(content=(
            "Write a comprehensive report on the current state of quantum computing "
            "in 2025, focusing on recent hardware breakthroughs and enterprise adoption."
        ))],
        "researcher_done": False,
        "fact_checker_done": False,
        "final_report": "",
    }

    result = graph.invoke(
        initial_state,
        config={"recursion_limit": 20}
    )

    print("=== FINAL REPORT ===")
    print(result["final_report"])
```

**Observable outcome:** What this demonstrates: - The supervisor inspects explicit state flags ( researcher done , fact checker done ) rather than relying on the LLM to infer completion — this makes routing deterministic and auditable.

## Stage 5 — langgraph-supervisor: The Official Multi-Agent Orchestration Library

- LangGraph ships a prebuilt library — langgraph-supervisor — that codifies the supervisor pattern with minimal boilerplate.
- Instead of writing the routing prompt and Command logic manually, create supervisor() generates it from your agent list.
- What create supervisor() does under the hood: 1.
- Builds a routing LLM prompt that includes each agent's name and description.

*Applies **llm-agents-and-langgraph** to the decision above.*

```bash
pip install langgraph-supervisor
```

```python
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# Create specialist agents using prebuilt ReAct loop
researcher_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="researcher",
    prompt="You are a research specialist. Find accurate, current information."
)

fact_checker_agent = create_react_agent(
    model=model,
    tools=[fact_check],
    name="fact_checker",
    prompt="You are a fact-checking specialist. Verify all claims carefully."
)

writer_agent = create_react_agent(
    model=model,
    tools=[],
    name="writer",
    prompt="You are a report writer. Synthesize findings into a clear report."
)

# Create the supervisor graph
supervisor = create_supervisor(
    agents=[researcher_agent, fact_checker_agent, writer_agent],
    model=model,
    prompt=(
        "You are a research team supervisor. Coordinate the researcher, "
        "fact_checker, and writer agents to produce accurate research reports. "
        "Always verify facts before writing the final report."
    )
).compile()

# Invoke
result = supervisor.invoke({
    "messages": [HumanMessage(content="Report on quantum computing in 2025")]
})
```

**Observable outcome:** Builds a routing LLM prompt that includes each agent's name and description.

## End-to-end outcome

After working through the stages, you should be able to explain:

- Make routing decisions observable, not opaque.
- The biggest debugging nightmare in multi-agent systems is a supervisor that routes inexplicably.

## Alternatives considered

### LangGraph supports three multi-agent coordination patterns. Each solves a different kind of decomposition problem.

LangGraph supports three multi-agent coordination patterns.

### Shared State and Worker Nodes

Every agent in the system reads from and writes to a shared TypedDict state.

### Supervisor Routing with Command

The supervisor node calls the LLM with a structured prompt that asks it to choose the next agent.

### Wiring the Graph

The graph topology is: supervisor routes to a worker → worker completes → supervisor evaluates → routes again or finishes.

### Decision Guide: Single Agent vs Supervisor vs Swarm

Decision Guide: Single Agent vs Supervisor vs Swarm


## Failure and degraded modes

### Performance vs. Coordination Cost

- **Detection:** Every inter-agent handoff adds at minimum one LLM call (the supervisor making a routing decision) plus state serialization overhead.
- **User or operator impact:** Every inter-agent handoff adds at minimum one LLM call (the supervisor making a routing decision) plus state serialization overhead.

### Agent Disagreement

- **Detection:** In a fact-checker + writer pair, the fact-checker may flag a claim as unverified while the writer proceeds to include it.
- **User or operator impact:** In a fact-checker + writer pair, the fact-checker may flag a claim as unverified while the writer proceeds to include it.

### Cascade Failures and Infinite Loops

- **Detection:** A supervisor that cannot decide emits the same routing decision repeatedly.
- **User or operator impact:** A supervisor that cannot decide emits the same routing decision repeatedly.

### Context Contamination

- **Detection:** Shared message history means every agent sees every other agent's outputs.
- **User or operator impact:** Shared message history means every agent sees every other agent's outputs.



## Scenario context

Swap InMemorySaver → PostgresSaver, add LangServe + Docker, trace with LangSmith.

<!--

Your LangGraph agent works perfectly in the demo. You deploy it to a single FastAPI instance. Two users hit it at the same time. They see each other's conversation history. The `InMemorySaver` you used in development doesn't know about concurrency — and it never will, because it was never designed for it.




This is the most co…

This walkthrough focuses on **The Demo-to-Production Gap: Why Notebook Agents Fail at Scale**.

Before making a design choice, clarify:

- Your LangGraph agent works perfectly in the demo.
- You deploy it to a single FastAPI instance.
- Two users hit it at the same time.

**Concepts in play:** llm-agents-and-langgraph, graph-traversal-algorithms, api-security-oauth2, api-gateway

## Stage 1 — LangServe: Wrapping a LangGraph Agent as a FastAPI App

- LangServe wraps any LangChain Runnable — including a compiled LangGraph — as a FastAPI application.
- Its core API is a single function: add routes .
- add routes auto-generates six endpoints from the graph's input/output schema: RemoteRunnable client — callers can use a typed Python client instead of raw HTTP: Limitation: LangServe is in maintenance mode as of 2025.
- directs new projects toward LangGraph Platform for managed deployments.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
# app.py
import os
from fastapi import FastAPI, Request, HTTPException
from langserve import add_routes
from agent import build_agent  # your compiled LangGraph

app = FastAPI(title="LangGraph Agent API", version="1.0")

# --- API key middleware ---
@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    if request.url.path.startswith("/agent"):
        key = request.headers.get("X-API-Key")
        if key != os.environ["AGENT_API_KEY"]:
            raise HTTPException(status_code=401, detail="Invalid API key")
    return await call_next(request)

# --- Mount the graph ---
agent = build_agent()

add_routes(
    app,
    agent,
    path="/agent",
    enable_feedback_endpoint=True,
    enable_public_trace_link_endpoint=True,
)

@app.get("/health")
async def health():
    return {"status": "ok"}
```

```python
from langserve import RemoteRunnable

agent = RemoteRunnable("http://localhost:8000/agent")

# Single turn
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Summarize Q4 results"}]},
    config={"configurable": {"thread_id": "user-42-session-7"}},
)

# Streaming
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Now compare to Q3"}]},
    config={"configurable": {"thread_id": "user-42-session-7"}},
):
    print(chunk)
```

**Observable outcome:** directs new projects toward LangGraph Platform for managed deployments.

## Stage 2 — Deep Dive: Concurrency, Persistence, and Production-Grade Checkpointing

- The Internals When a POST /agent/invoke request arrives at a LangServe worker, here is the exact execution path: 1.
- Deserialize — LangServe uses the graph's InputType schema to parse the JSON body.
- Config extraction — the config.configurable.thread id is extracted and passed to the checkpointer.
- Checkpoint read — AsyncPostgresSaver opens a connection from its asyncpg pool and runs a SELECT for the latest checkpoint row matching thread id .

*Applies **graph-traversal-algorithms** to the decision above.*

```
Worker 1 memory: { "user-42": [msg1, msg2] }
Worker 2 memory: { }           ← empty, knows nothing about msg1 or msg2

User's turn 3 routes to Worker 2 → fresh conversation, no history
```

```python
# Development (single process only)
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

# Production (multi-worker safe)
import asyncpg
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def build_production_graph():
    pool = await asyncpg.create_pool(dsn=os.environ["DATABASE_URL"], min_size=2, max_size=10)
    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()   # idempotent: creates tables if absent
    return builder.compile(checkpointer=checkpointer)
```

**Observable outcome:** Checkpoint read — AsyncPostgresSaver opens a connection from its asyncpg pool and runs a SELECT for the latest checkpoint row matching thread id .

## Stage 3 — Beyond Request-Response: Auth Patterns, Background Runs, and LangGraph Platform

- Authentication and Per-User Thread Isolation LangServe provides no authentication out of the box — everything added in the middleware snippet earlier is your responsibility.
- The three common production patterns are: Pattern 1 — Static API key (internal tools and service-to-service) Simple, effective for a single service identity.
- Does not support per-user tracking or revocation.
- Pattern 2 — JWT with per-user thread isolation The thread id construction here is the security boundary.

*Applies **api-security-oauth2** to the decision above.*

```python
# In FastAPI middleware — shown earlier in ⚙️ section
key = request.headers.get("X-API-Key")
if key != os.environ["AGENT_API_KEY"]:
    raise HTTPException(status_code=401)
```

```python
import jwt
from fastapi import Depends, Request

SECRET = os.environ["JWT_SECRET"]

async def get_current_user(request: Request) -> dict:
    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return payload  # contains {"sub": "user-42", "tenant": "acme"}

@app.post("/agent/invoke")
async def invoke(body: dict, user: dict = Depends(get_current_user)):
    thread_id = f"{user['tenant']}-{user['sub']}-{body['session_id']}"
    return await graph.ainvoke(
        body["input"],
        config={"configurable": {"thread_id": thread_id}},
    )
```

**Observable outcome:** Pattern 2 — JWT with per-user thread isolation The thread id construction here is the security boundary.

## Stage 4 — Production Architecture: From Load Balancer to LangSmith

- The following diagram shows the full request path for a multi-worker LangServe deployment with PostgreSQL persistence and LangSmith observability.
- The dashed arrows are async/non-blocking paths that do not add to user-facing latency.
- Enabling Tracing What is captured per run: Reading a Trace to Debug a Failed Run 1.
- Open the LangSmith dashboard → select your project.

*Applies **api-gateway** to the decision above.*

```bash
# .env (or container environment)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...            # from app.langsmith.com
LANGCHAIN_PROJECT=react-agent-prod  # project name in LangSmith dashboard
```

```
Example trace tree for a failed ReAct run:

  ✅ graph.invoke (1.8s)
    ✅ node: classify_intent (0.3s)
    ✅ node: search_tool (0.6s)
    ❌ node: format_response (0.2s)
         ↳ KeyError: 'product_name' not found in tool output
         ↳ Input state: {"search_results": [], "messages": [...]}
```

**Observable outcome:** Open the LangSmith dashboard → select your project.

## End-to-end outcome

After working through the stages, you should be able to explain:

- InMemorySaver is a deployment hazard, not just a development convenience.
- It works in every test, fails silently in every multi-worker deploy.

## Alternatives considered

### Deployment Options Compared: LangServe vs LangGraph Platform vs Self-Hosted

Before writing a single line of deployment code, choose the right layer.


## Failure and degraded modes

The source does not isolate a separate failure mode. Limits that matter for the scenario are called out in the stage decisions above.

## Scenario context

A tool is a single callable capability (search, SQL, calculator). A skill is a reusable mini-workflow that coordinates multiple tool calls with policy, guardrails, retries, and output structure. If you model everything as "just tools," your agent usually works in demos but fails in production.

---

An AI agent built to assist financial analysts had a `calculator` tool registered in its tool list. Yet in testing, it…

This walkthrough focuses on **Why "Skill" Is Not Just a Fancy Name for "Tool"**.

Before making a design choice, clarify:

- An AI agent built to assist financial analysts had a calculator tool registered in its tool list.
- Yet in testing, it kept producing arithmetic errors — computing 12.4 8.7 and returning 107.9 instead of the correct 107.88 .
- The calculator tool itself worked perfectly when called.

**Concepts in play:** llm-agents-and-langgraph, event-driven-architecture, graph-traversal-algorithms, sorting-and-searching-algorithms

## Stage 1 — The Three-Layer Mental Model: Model, Tools, Skills

- A practical way to design modern agents is with three layers: 1.
- LLM layer : reasoning, planning, and language generation.
- Tool layer : external operations (APIs, databases, code execution, search).
- Skill layer : orchestrated routines that solve recurring goals.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Skill layer : orchestrated routines that solve recurring goals.

## Stage 2 — How a Skill Actually Runs Across Multiple Tools

- Suppose the user asks: "Investigate this outage alert and open a ticket with a clear summary." A tool-only design might call APIs opportunistically.
- A skill-based design follows a known contract.
- Validate input schema ( service , time range , alert id ).
- Run consistency checks (for example, severity must match evidence).

*Applies **event-driven-architecture** to the decision above.*

```mermaid
flowchart TD
    A[User goal: investigate outage and open ticket] --> B[Planner selects IncidentTriageSkill]
    B --> C[Step 1: fetch logs tool]
    C --> D[Step 2: classify severity tool]
    D --> E[Step 3: summarize findings tool]
    E --> F[Step 4: create ticket tool]
    F --> G[Return structured result: summary, severity, ticket_id]
```

**Observable outcome:** Run consistency checks (for example, severity must match evidence).

## Stage 3 — Deep Dive: What Makes a Skill Reliable in Production

- The internals: a skill is policy plus orchestration A production-grade skill usually includes these internal parts: This architecture lets you debug behavior at the skill level instead of reverse-engineering long prompt…
- Mathematical model: choosing the best skill for a goal When several skills could solve a request, use an explicit routing score: $$ Score(skill i \mid goal) = \alpha C i - \beta L i - \gamma R i + \delta F i $$ Where: -…
- This is not "academic math." It is a practical routing heuristic that prevents random skill selection.
- Performance analysis: skills add overhead but reduce incident rate Skills trade a little raw speed for much better reliability and operator confidence.

*Applies **graph-traversal-algorithms** to the decision above.*

**Observable outcome:** Performance analysis: skills add overhead but reduce incident rate Skills trade a little raw speed for much better reliability and operator confidence.

## Stage 4 — Real-World Application Patterns

- Case study 1: Support triage assistant - Input : incoming ticket text and account metadata.
- - Process : skill calls sentiment tool, policy lookup tool, and routing API.
- - Output : priority, queue assignment, and draft response.
- Case study 2: Engineering incident assistant - Input : alert payload from monitoring system.

*Applies **sorting-and-searching-algorithms** to the decision above.*

**Observable outcome:** Case study 2: Engineering incident assistant - Input : alert payload from monitoring system.

## Stage 5 — Practical Examples: Implementing a Skill Layer

- Example 1: Declare tools and a skill contract These examples build a skill layer incrementally: first by declaring individual tool functions and a typed input contract, then by wrapping them with retry logic, confidence…
- This progression was chosen to demonstrate that skill design is an architectural discipline, not a framework choice — the same pattern applies in pure Python before any orchestration library is introduced.
- When reading the code, focus on how the output shape and error handling are specified inside the skill function itself: this is what separates a skill runtime from a simple function call chain.
- This is already more robust than free-form tool hopping because the output shape is stable.

*Applies **llm-agents-and-langgraph** to the decision above.*

```python
from dataclasses import dataclass
from typing import Any, Dict

def fetch_logs(service: str, time_range: str) -> str:
    # Placeholder for real API integration.
    return f"logs(service={service}, window={time_range})"

def classify_severity(log_blob: str) -> Dict[str, Any]:
    return {"severity": "high", "confidence": 0.87}

def create_ticket(summary: str, severity: str) -> str:
    return "INC-48291"

@dataclass
class IncidentInput:
    service: str
    time_range: str
    alert_id: str

def incident_triage_skill(payload: IncidentInput) -> Dict[str, Any]:
    logs = fetch_logs(payload.service, payload.time_range)
    cls = classify_severity(logs)
    summary = f"Alert {payload.alert_id} appears {cls['severity']} severity"
    ticket_id = create_ticket(summary, cls["severity"])
    return {
        "summary": summary,
        "severity": cls["severity"],
        "confidence": cls["confidence"],
        "ticket_id": ticket_id,
    }
```

```python
import time

def run_with_retry(fn, max_attempts=3, base_delay=0.5):
    for attempt in range(1, max_attempts + 1):
        try:
            return fn()
        except Exception:
            if attempt == max_attempts:
                raise
            time.sleep(base_delay * attempt)

def safe_incident_triage(payload: IncidentInput) -> Dict[str, Any]:
    if not payload.service or not payload.time_range:
        raise ValueError("service and time_range are required")

    logs = run_with_retry(lambda: fetch_logs(payload.service, payload.time_range))
    cls = run_with_retry(lambda: classify_severity(logs))

    if cls["confidence"] < 0.60:
        return {
            "status": "needs_human_review",
            "reason": "low_classifier_confidence",
            "alert_id": payload.alert_id,
        }

    summary = f"Alert {payload.alert_id} appears {cls['severity']} severity"
    ticket_id = run_with_retry(lambda: create_ticket(summary, cls["severity"]))

    return {
        "status": "ok",
        "summary": summary,
        "severity": cls["severity"],
        "ticket_id": ticket_id,
    }
```

**Observable outcome:** This is already more robust than free-form tool hopping because the output shape is stable.

## Stage 6 — LangChain Tools API: Registering Atomic Tools and Promoting Them to Skills

- LangChain provides Tool and StructuredTool abstractions that formalize the tool layer described throughout this post.
- StructuredTool adds a Pydantic input schema, making tool calls type-safe, self-documenting, and enabling the LLM to validate inputs before execution — the exact boundary that separates reliable tool use from ad-hoc prom…
- StructuredTool wraps the entire skill — including its internal multi-step tool orchestration — behind a single schema-validated interface.
- The LLM calls one tool; internally it runs the sequence, applies the policy gate, and returns a structured result.

*Applies **event-driven-architecture** to the decision above.*

```python
# pip install langchain langchain-core pydantic
from langchain_core.tools import Tool, StructuredTool
from pydantic import BaseModel, Field
from typing import Any, Dict

# --- Layer 1: Simple Tool — one string input, minimal guardrails ---
fetch_logs_tool = Tool(
    name="fetch_logs",
    description="Fetch recent log snippets for a service name. Returns a raw log string.",
    func=lambda service: f"[ERROR] 3 timeout errors in {service} over last 15 min",
)

# --- Layer 2: StructuredTool — typed multi-field input with Pydantic validation ---
class IncidentInput(BaseModel):
    service:    str = Field(description="Service name, e.g. 'payments-svc'")
    time_range: str = Field(description="Lookback window, e.g. 'last_15m'")
    alert_id:   str = Field(description="Unique alert ID from monitoring system")

def triage_incident(service: str, time_range: str, alert_id: str) -> Dict[str, Any]:
    """
    Skill function: orchestrates sub-tools internally.
    The LLM sees one tool; internally it runs multiple steps with built-in policy.
    """
    logs     = fetch_logs_tool.run(service)
    severity = "high" if "error" in logs.lower() else "low"

    # Policy gate: only escalate confirmed high-severity alerts
    if severity == "high" and "timeout" in logs:
        ticket_id = f"INC-{abs(hash(alert_id)) % 9999:04d}"
    else:
        ticket_id = None

    return {
        "alert_id":  alert_id,
        "severity":  severity,
        "ticket_id": ticket_id,
        "summary":   f"Service {service} shows {severity} severity over {time_range}",
    }

incident_skill = StructuredTool.from_function(
    func=triage_incident,
    name="incident_triage_skill",
    description=(
        "Run full incident triage: fetch logs, classify severity, open ticket if high. "
        "Returns structured report with alert_id, severity, ticket_id, and summary."
    ),
    args_schema=IncidentInput,
)

# The LLM calls one structured tool — the skill handles all internal orchestration
print(incident_skill.invoke({
    "service": "payments-svc",
    "time_range": "last_15m",
    "alert_id": "ALT-9910",
}))
```

**Observable outcome:** The LLM calls one tool; internally it runs the sequence, applies the policy gate, and returns a structured result.

## End-to-end outcome

After working through the stages, you should be able to explain:

- - Treat tools as primitives, not products.
- Skills are where product behavior actually lives.

## Alternatives considered

### Control-Flow View: Single Tool Call vs Skill Runtime

A side-by-side sequence perspective makes the distinction obvious.

### Decision Guide: Should This Be a Tool, a Skill, or a Workflow Engine?

Use this heuristic: if your prompt keeps repeating the same sequence of tool calls, promote that sequence into a skill.


## Failure and degraded modes

### Trade-offs & Failure Modes: Trade-offs and Failure Modes You Should Plan For

- **Detection:** They add a control layer, and that layer must be designed carefully.
- **User or operator impact:** They add a control layer, and that layer must be designed carefully.



## Scenario context

If tools are primitives and skills are reusable routines, then the **skill registry + router + evaluator** is your production control plane. This layer decides which skill runs, under what constraints, and how you detect regressions before users do.

---

In small demos, the model picks a tool and returns a decent answer. In production, that is not enough.




You need answers to operational questions:

- Which skil…

This walkthrough focuses on **Why a Skill Registry Becomes the Agent Control Plane**.

Before making a design choice, clarify:

- In small demos, the model picks a tool and returns a decent answer.
- In production, that is not enough.
- You need answers to operational questions: - Which skills are currently active?

**Concepts in play:** llm-agents-and-langgraph, vector-embeddings-and-similarity, database-indexing, consistency-replication

## Stage 1 — Designing a Registry That Humans and Routers Can Trust

- A useful registry entry is both machine-readable and operator-readable.
- Minimum fields per skill: Dense systems should also include: - deprecation status, - fallback skill id, - required approvals (for sensitive actions), - evaluation baseline hash.
- A registry is a product artifact.
- Treat it like API surface area, not internal trivia.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** Treat it like API surface area, not internal trivia.

## Stage 2 — Routing Pipeline: From User Intent to Skill Selection

- A production router should be explicit about stages.
- This pipeline prevents a common failure mode: the model picks a "kind of related" skill because a keyword looked similar.
- Router quality is usually more important than incremental prompt tweaks once you scale beyond a few skills.

*Applies **vector-embeddings-and-similarity** to the decision above.*

```mermaid
flowchart TD
    A[User request] --> B[Intent and entity extraction]
    B --> C[Candidate skill retrieval from registry]
    C --> D[Policy gate: data/risk/compliance]
    D --> E[Score candidates: fit, cost, risk, freshness]
    E --> F{Confidence above threshold?}
    F -- Yes --> G[Select top skill]
    F -- No --> H[Fallback: safe default or human review]
    G --> I[Execute skill with trace]
    H --> I
    I --> J[Return response + route metadata]
```

**Observable outcome:** Router quality is usually more important than incremental prompt tweaks once you scale beyond a few skills.

## Stage 3 — Deep Dive: Scoring, Constraints, and Runtime Guarantees

- Internals: hybrid routing usually beats single-strategy routing Most robust systems combine three routing signals: 1.
- Rule-based filters for non-negotiable constraints (risk, permissions, domain).
- Semantic match for intent-to-skill relevance.
- Operational priors from latency, error rate, and freshness.

*Applies **database-indexing** to the decision above.*

**Observable outcome:** Operational priors from latency, error rate, and freshness.

## Stage 4 — Evaluation Loop: Offline Replay, Shadow Routing, and Live Gates

- Recommended evaluation layers: For intermediate maturity, start with one robust offline suite and one shadow dashboard before touching canary automation.

*Applies **consistency-replication** to the decision above.*

```mermaid
sequenceDiagram
    participant D as Dataset Store
    participant R as Router
    participant E as Evaluator
    participant P as Prod Traffic

    D->>R: historical requests replay
    R-->>E: selected skill + confidence + trace
    E->>E: compute quality/safety/latency metrics
    E-->>R: threshold updates and alerts
    P->>R: live requests (shadow mode)
    R-->>E: shadow route decisions
    E-->>R: promote or rollback recommendation
```

**Observable outcome:** Recommended evaluation layers: For intermediate maturity, start with one robust offline suite and one shadow dashboard before touching canary automation.

## Stage 5 — Real-World Applications: Rollout Patterns That Work in Real Teams

- Pattern 1: New skill onboarding checklist - Add skill metadata and policy fields to registry.
- - Add at least 20 representative replay prompts.
- - Verify schema-valid output rate and safety checks.
- - Enable shadow routing before any user-facing traffic.

*Applies **llm-agents-and-langgraph** to the decision above.*

**Observable outcome:** - Enable shadow routing before any user-facing traffic.

## Stage 6 — Practical Examples: Registry and Router Skeleton

- Example 1: Registry document shape (JSON) These examples provide a concrete skill registry document shape and a hybrid route selection function that together form the skeleton of a production routing system.
- The JSON registry entry was chosen because it is the minimum viable artifact that makes routing decisions both machine-readable and operator-auditable — humans and automated routers share the same source of truth.
- When reading the Python routing function, focus on how eligibility filtering happens before score computation: this separation is what keeps policy enforcement from leaking into the scoring math and makes each layer ind…
- This is enough to power both routing decisions and operator dashboards.

*Applies **vector-embeddings-and-similarity** to the decision above.*

```json
{
  "skill_id": "incident_triage_v3",
  "version": "3.2.1",
  "description": "Analyze outage alerts, summarize impact, and create incident tickets.",
  "risk_level": "medium",
  "owner_team": "sre-platform",
  "allowed_data_domains": ["logs", "incidents"],
  "input_schema": {
    "type": "object",
    "required": ["service", "time_range", "alert_id"]
  },
  "output_schema": {
    "type": "object",
    "required": ["summary", "severity", "ticket_id"]
  },
  "slo": {
    "p95_latency_ms": 4000,
    "schema_valid_rate": 0.99
  },
  "fallback_skill_id": "incident_triage_safe_v1"
}
```

```python
from typing import Dict, List

def allowed(skill: Dict, request: Dict) -> bool:
    if request.get("risk_tier") == "high" and skill.get("risk_level") == "high":
        return False
    required_domain = request.get("domain")
    return required_domain in skill.get("allowed_data_domains", [])

def route_score(skill: Dict, fit: float, latency_ms: float, reliability: float) -> float:
    risk_penalty = {"low": 0.05, "medium": 0.15, "high": 0.40}[skill.get("risk_level", "medium")]
    return 0.55 * fit - 0.20 * (latency_ms / 5000.0) - 0.15 * risk_penalty + 0.10 * reliability

def choose_skill(request: Dict, candidates: List[Dict]) -> Dict:
    scored = []
    for skill in candidates:
        if not allowed(skill, request):
            continue

        # Placeholder signals. In production these come from embedding match and telemetry.
        fit = skill.get("fit", 0.0)
        latency_ms = skill.get("p95_latency_ms", 3000)
        reliability = skill.get("schema_valid_rate", 0.95)

        scored.append((route_score(skill, fit, latency_ms, reliability), skill))

    if not scored:
        return {"action": "fallback", "reason": "no_allowed_candidate"}

    scored.sort(key=lambda pair: pair[0], reverse=True)
    best_score, best_skill = scored[0]

    if best_score < 0.35:
        return {"action": "fallback", "reason": "low_confidence", "score": best_score}

    return {"action": "execute", "skill_id": best_skill["skill_id"], "score": best_score}
```

**Observable outcome:** This is enough to power both routing decisions and operator dashboards.

## End-to-end outcome

After working through the stages, you should be able to explain:

- - Registry design quality determines routing quality more than people expect.
- - If metadata ownership is unclear, incident resolution slows down dramatically.

## Alternatives considered

### Decision Guide: What to Build First at Your Current Maturity

Decision Guide: What to Build First at Your Current Maturity


## Failure and degraded modes

### Trade-offs & Failure Modes: Failure Modes and Mitigations in Skill Routing Systems

- **Detection:** Also watch for this anti-pattern: using one global confidence threshold for every intent type.
- **User or operator impact:** Also watch for this anti-pattern: using one global confidence threshold for every intent type.



Scale agents with memory, multi-agent coordination, deployment, observability, and skill routing.

Production Agent Systems

Build stateful agents with tools, planning, LangGraph, memory, human-in-the-loop workflows, multi-agent coordination, and production architectures.