How to build AI agents using LangChain

An AI agent is an autonomous system that uses an LLM as its reasoning engine to decide which actions to take. Unlike a chatbot that answers a prompt and stops, an agent uses tools, reaches external data, and chains several steps together to finish a task it was never given step-by-step instructions for.

This is the build-along: from an empty file to an agent that runs a multi-step workflow, using LangChain for the loop and LangGraph for the graph. It goes past the part every tutorial covers — the first working invoke() — into the part that decides whether the thing survives a week in production: tool design, state, failure routing, cost ceilings, evals and approval gates.

#What an agent actually is

Strip the vocabulary away and an agent is one loop.

You call the model with a task and a list of tools it is allowed to use. The model either answers, or it asks for a tool call. Your code runs the tool, appends the result to the conversation, and calls the model again — with the new information in context. The model, not your code, decides whether to go round again.

That inversion is the entire distinction. A chain has a control flow you wrote. An agent has a control flow the model chooses at runtime, inside limits you set. Everything difficult about building agents follows from that one sentence: you're not debugging a program, you're constraining a decision-maker.

#The three moving parts

A LangChain agent has three:

• LLM (the reasoning engine) — decides what to do next
• Tools — the functions it is allowed to call: search, code execution, APIs
• Memory — conversation and state that survives across steps

Everything else is plumbing around those three. Most of the quality difference between a demo and a production agent lives in the second one.

#Setting up

pip install langchain langgraph langchain-openai tavily-python

Set the keys your provider and tools need — OPENAI_API_KEY, TAVILY_API_KEY — as environment variables rather than literals. An agent's traces get shipped to observability platforms, and a key pasted into a prompt template ends up in them.

#Your first agent

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
 
# The reasoning engine. temperature=0 for anything that has to be repeatable.
llm = ChatOpenAI(model="gpt-4o", temperature=0)
 
# The functions the model is allowed to call.
tools = [search_tool, calculator_tool, code_executor]
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a research assistant. Use tools for anything factual. "
               "If a tool fails twice, say so and stop — do not guess."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
 
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=8,        # hard ceiling on loops
    max_execution_time=90,   # hard ceiling on wall clock
    return_intermediate_steps=True,
)
 
result = executor.invoke({"input": "Analyze NVIDIA's revenue trend since 2023"})

agent_scratchpad is the loop made visible: every tool call and every result gets appended there, and the whole growing transcript is what the model reads on the next pass. max_iterations and max_execution_time are not optional polish. Without them the failure mode of a confused agent is an infinite loop billed per token.

That is where most tutorials end. It works, and it is roughly a tenth of the job.

#Tools are the part that decides quality

Teams spend their time tuning the system prompt when the tool layer is what's actually failing. A tool definition is three things to the model — a name, a description and an argument schema — and it can only be as reliable as those.

from langchain_core.tools import tool
from pydantic import BaseModel, Field
 
class RevenueQuery(BaseModel):
    ticker: str = Field(description="Exchange ticker, uppercase, e.g. NVDA")
    start_year: int = Field(description="First fiscal year, inclusive", ge=2015)
    end_year: int = Field(description="Last fiscal year, inclusive")
 
@tool("quarterly_revenue", args_schema=RevenueQuery)
def quarterly_revenue(ticker: str, start_year: int, end_year: int) -> dict:
    """Quarterly reported revenue in USD millions for one listed company.
 
    Use for revenue only — not margins, headcount or share price.
    Returns {"rows": [...]} on success, or {"error": "..."} when the ticker is
    unknown or the range has no filings. Safe to call repeatedly.
    """
    try:
        rows = filings.revenue(ticker, start_year, end_year)
    except UnknownTicker:
        return {"error": f"No filings for ticker {ticker}. Check the symbol."}
    if not rows:
        return {"error": f"No filings for {ticker} between {start_year} and {end_year}."}
    return {"rows": rows, "unit": "USD millions"}

Four things there are doing real work:

• The docstring is a prompt, not documentation. It's the only thing the model reads when deciding whether this tool fits. Say what it's for and what it isn't — half of all wrong tool calls are a model reaching for the closest thing available.
• The schema constrains before the call happens. ge=2015 rejects a hallucinated 1970 at the boundary instead of returning a confusing empty result the model then reasons over.
• Failures come back as data, not exceptions. A raised exception either kills the run or gets stringified into the context as a stack trace. A {"error": "..."} the model can read is a fact it can act on — usually by correcting the argument and retrying, which is exactly what you want.
• Idempotency is stated. Agents retry. A tool that charges a card or sends an email needs an idempotency key, not a hopeful docstring.

#Memory and state are different things

"Memory" gets used for two things that behave nothing alike, and conflating them is how agents end up both forgetful and expensive.

Conversation context is the transcript the model sees on this call. It is bounded by the context window, it costs tokens on every single pass, and it grows with each tool result. Left unmanaged, a ten-step agent re-reads nine irrelevant tool outputs on step ten. Trim it: keep the task, the last few exchanges and a running summary; drop raw tool payloads once their conclusion has been extracted.

Durable state is what survives the run — what was decided, what was produced, what has already been attempted. It belongs in a store you control, keyed by thread, not in the transcript. The practical test: if the process crashed right now, what would you need to resume? That is state. Everything else is context.

#Where the single loop breaks

The symptoms are consistent. The agent redoes work it already did. It skips a step under a long transcript. Two runs on the same input take different paths. Adding an instruction to fix one behaviour breaks another. You are trying to express a workflow in prose, and prose has no guarantees.

#LangGraph: the work becomes a graph

LangGraph extends LangChain with stateful, multi-actor workflows. Instead of one linear chain, you define a graph: each node is an agent or a function, edges define control flow, and every node reads and writes one shared, typed state object.

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
 
class ResearchState(TypedDict):
    task: str
    findings: Annotated[list[str], add]  # nodes append; the reducer merges
    draft: str
    grade: str
    attempts: int
 
def grade_route(state: ResearchState) -> str:
    """Conditional edge: the return value names the next node."""
    if state["grade"] == "pass":
        return "write"
    if state["attempts"] >= 2:
        return "escalate"     # bounded — never loop on quality forever
    return "analyze"
 
workflow = StateGraph(ResearchState)
workflow.add_node("web", web_research)
workflow.add_node("docs", internal_docs)
workflow.add_node("analyze", analyst)
workflow.add_node("write", writer)
workflow.add_node("escalate", ask_a_human)
 
# Fan-out: both retrieval nodes run in parallel from START.
workflow.add_edge(START, "web")
workflow.add_edge(START, "docs")
# Fan-in: analyze waits for both, then the reducer has merged their findings.
workflow.add_edge("web", "analyze")
workflow.add_edge("docs", "analyze")
workflow.add_conditional_edges("analyze", grade_route)
workflow.add_edge("write", END)
 
app = workflow.compile()
result = app.invoke({"task": "GPU supply, 2026", "attempts": 0})

Three things changed, and none of them are about prompting:

1. Parallelism is declared. Two edges out of START means both retrieval nodes run concurrently. The Annotated[list[str], add] reducer is what makes the fan-in safe — without it, concurrent writes to the same key are a conflict, not a merge.
2. Branching is code. grade_route is a plain Python function. It is readable, testable and identical on every run.
3. The loop is bounded by a counter, not by a sentence asking the model to please stop.

Once the work is a graph, the interesting questions stop being about prompting. They become: what runs in parallel, where does a failed step route, and which gate has to pass before anything ships.

#Routing failure

Every node is a place something can fail, and the three failure modes want different handling.

Transient failures — a timeout, a 503, a rate limit — are a retry policy on the node, not a decision for the model to reason about:

from langgraph.pregel import RetryPolicy
 
workflow.add_node(
    "web",
    web_research,
    retry=RetryPolicy(max_attempts=3, initial_interval=0.5, backoff_factor=2),
)

Bad output — the model produced something that doesn't validate — routes to a repair node with the validation error attached, so the second attempt is informed rather than hopeful:

def validate(state: ResearchState) -> dict:
    try:
        Report.model_validate_json(state["draft"])
        return {"grade": "pass"}
    except ValidationError as e:
        # The error text goes back into state; the repair node reads it.
        return {"grade": "fail", "findings": [f"validation: {e.errors()}"],
                "attempts": state["attempts"] + 1}

Genuine dead ends — the data doesn't exist, the request is out of scope — route to a human. An agent that cannot fail loudly will fail quietly, and a confident wrong answer costs more than a stopped run.

#Cost and latency

Agents fail commercially before they fail technically. Three controls, in order of how much they return:

• Budget per run, not per month. A monthly dashboard tells you about the incident after it happened. Count tokens inside the run and hard-stop the graph when it crosses the ceiling — the same way max_iterations works, but denominated in money.
• Tier the models per node. Routing, grading, classification and extraction do not need your most expensive model. Reserve it for the nodes that actually reason. On a six-node graph this is routinely a 60–80% cost reduction with no measurable quality change — but measure it, per node, before assuming.
• Cache and trim. Cache tool results within a thread; a well-designed idempotent tool makes this trivial. Trim the transcript as described above — in a long agent run, resent context is usually the single largest line item.

#Observability and evals

You cannot debug an agent from its final answer. The unit of debugging is the trace: every model call, every tool call, every argument, every state transition, with timings and token counts. LangSmith does this natively; OpenTelemetry-based stacks work fine too. The requirement is not the vendor, it is that a failed run can be replayed step by step.

Evals matter more, and they only pay off when they are a gate rather than a report. A number in a dashboard nobody blocks on changes nothing. Run the suite on every prompt, model and tool change, and let a regression stop the deploy.

Keep the suite small and real — thirty cases drawn from actual failures beat three hundred synthetic ones. Every production bug becomes a case. That is how the suite stays honest.

#Human-in-the-loop

Some steps should not be automated, and LangGraph makes the pause a first-class part of the graph rather than a hack:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
 
def approve(state: ResearchState) -> dict:
    decision = interrupt({"draft": state["draft"]})   # pauses; state persists
    if decision["approved"]:
        return {"grade": "pass"}
    return {"grade": "fail", "findings": [decision["reason"]]}
 
# A checkpointer is what makes the pause survivable — use a real store
# (Postgres, Redis) in production; MemorySaver is for local runs.
app = workflow.compile(checkpointer=MemorySaver())

The run stops, the state is checkpointed, and it resumes when a human answers — possibly days later, in a different process. Put a gate anywhere the action is irreversible, expensive, or externally visible: money moving, code merging, messages sent to customers. Everywhere else, the friction is not worth it.

#At delivery scale

The same structure holds when the work is not one report but a stream of tickets. Agents inside sandboxes, status gates between phases, pass/fail edges deciding what moves forward, and a human on the last edge before production:

Nothing there is a different technique from the graph above. It is the same five ideas — typed state, bounded loops, validated output, gates, an approval edge — applied to a bigger unit of work.

#The checklist

Before an agent runs unattended:

• Bounded — iteration cap, wall-clock cap, token budget per run
• Validated — every tool output and every final answer checked against a schema before it moves on
• Sandboxed — no production credentials in the loop; write access mediated by a tool that can be audited
• Traced — a failed run can be replayed call by call
• Gated — evals block the deploy; a human blocks the irreversible action
• Reversible — every side effect can be undone, or is idempotent enough that repeating it is harmless

#Where this goes

The future is not a single agent with a longer tool list. It is agent ecosystems — specialized agents that collaborate, delegate and self-correct, with the harness around them deciding what is allowed to happen.

That harness — the graph, the gates, the evals — is the part that turns a probabilistic text generator into a repeatable production system. The model is the commodity. The structure you put around it is the engineering.