03 - LangChain and LangGraph: Agents with AgentExecutor and Tools

The problem: LangChain tutorials no longer run

If you copy a LangChain agent example written a year ago, it probably starts with from langchain.agents import AgentExecutor and today it stops with an import error. It is not your mistake: the ecosystem has been reorganized. The short answer to "how do I build an agent today" is: with create_agent from the langchain package if the ready-made loop is enough for you, with a StateGraph from LangGraph if you want to control the steps, the saved states and the pauses for human approval yourself. This article shows both routes, and so as not to depend on a paid service it uses a fake model that follows a fixed rule: the listings are really executed, but they test the structure, not the intelligence of a model.

First of all, the measurement. The following block tries to import AgentExecutor from langchain.agents (the route of the old tutorials) and then from the langchain-classic package, and prints the versions installed in a clean virtual environment. The examples in the article were run with Python 3.14.4 and these versions, installed from PyPI on 2026-11-12.

import importlib.metadata as md

try:
    from langchain.agents import AgentExecutor
    print("langchain.agents.AgentExecutor: imported")
except ImportError as e:
    print("langchain.agents.AgentExecutor:", type(e).__name__, "from", e.name)

from langchain_classic.agents import AgentExecutor
print("langchain_classic.agents.AgentExecutor:", AgentExecutor.__module__)
for package in ("langchain", "langchain-classic", "langgraph"):
    print(package, md.version(package))

# Output:
# langchain.agents.AgentExecutor: ImportError from langchain.agents
# langchain_classic.agents.AgentExecutor: langchain_classic.agents.agent
# langchain 1.5.0
# langchain-classic 1.0.8
# langgraph 1.2.21

With langchain 1.5.0 the import from the old location fails with ImportError, while langchain-classic 1.0.8 still exposes the class. The official migration guide to version 1 says that the legacy code of the langchain package (chains such as LLMChain, retrievers, indexing, the hub, deprecated features) has been moved to langchain-classic, and that in place of langgraph.prebuilt.create_react_agent it recommends langchain.agents.create_agent. We did not run a complete loop of AgentExecutor: the block only tests where the class lives. For new code there is no reason to start from there.

The packages: who does what

The table explains why versions matter: the numbers are not aligned across packages (langgraph 1.2.21, langchain 1.5.0, langchain-classic 1.0.8, langchain-ollama 1.1.0, langchain-openai 1.7.4, langchain-anthropic 1.7.10 as of 2026-11-12), and an example that does not declare which ones it installs is fragile. Integrations are separate packages: if you use a local model served by Ollama, the package is langchain-ollama, not the old import from langchain-community that appears in tutorials. Naming a provider here is a matter of compatibility: the article presupposes none.

The fake model, and why we use it

To test an agent without calling a model, a subclass of BaseChatModel with two methods is enough: bind_tools, which here returns itself because the fake model does not read schemas, and _generate, which decides the reply by looking at the last message. If the last one is a tool result (ToolMessage), it answers with a final sentence; otherwise it proposes the call to find_order with the argument A-1042. It is the same form of exchange as a real model (request, tool call, result, answer), but without reasoning: we will use it only to check that the wiring works. It does not prove that a real model would choose the right tool.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class FakeModel(BaseChatModel):
    """Behaves like a model with tools, but follows a fixed rule:
    after a question it calls the tool; after the result it answers."""

    @property
    def _llm_type(self) -> str:
        return "fake"

    def bind_tools(self, tools, **kwargs):
        return self                      # the fake model does not need the schemas

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        last = messages[-1]
        if isinstance(last, ToolMessage):
            msg = AIMessage(content=f"Final answer: {last.content}")
        else:
            msg = AIMessage(content="", tool_calls=[
                {"name": "find_order", "args": {"id": "A-1042"}, "id": "call_1"}])
        return ChatResult(generations=[ChatGeneration(message=msg)])

The file is called fake_model.py and is imported by the blocks that follow; it prints nothing. One detail that matters: the id field in the call (call_1) is mandatory, because it is used to match the result to the request; if you omit it, execution stops with TypeError: tool_call() missing 1 required keyword-only argument: 'id' (tested).

The tools: name, description, schema

The @tool decorator from langchain_core.tools turns a function into a tool: the name comes from the function, the description from the docstring, the argument schema from the type hints. The model sees only these three elements, so the docstring is a prompt for all intents and purposes: it says when to use the tool and when not to. The block uses a dictionary in place of a search service, to stay runnable offline, and prints what the model would see, then invokes the tool with correct arguments and with a wrong argument name.

from langchain_core.tools import tool

ORDERS = {"A-1042": "out for delivery", "A-1043": "shipped"}

@tool
def find_order(id: str) -> str:
    """Returns the status of an order given its code (format A-NNNN).
    Use it for questions about the status of an order, not for prices or returns."""
    return ORDERS.get(id, "order does not exist")

print(find_order.name)
print(find_order.description)
print(find_order.args)
print(find_order.invoke({"id": "A-1042"}))
try:
    find_order.invoke({"identifier": "A-1042"})
except Exception as e:
    print(type(e).__name__, "-", str(e).splitlines()[0])

# Output:
# find_order
# Returns the status of an order given its code (format A-NNNN).
# Use it for questions about the status of an order, not for prices or returns.
# {'id': {'title': 'Id', 'type': 'string'}}
# out for delivery
# ValidationError - 1 validation error for find_order

The output shows the generated schema ({'id': {'title': 'Id', 'type': 'string'}}) and the fact that a call with the argument identifier instead of id does not reach the function: Pydantic's validation blocks it with ValidationError. In a real agent that error must go back to the model as text, otherwise it cannot correct itself. If you remove the str type hint, the schema loses the type and the model receives fewer indications about how to fill in the field.

Tools that touch data: filtering the text is not enough

A frequent example in tutorials is a query_database tool that accepts SQL and runs it "only if it starts with SELECT". That check on the text is not a barrier: real protection is placed where writing is impossible. With SQLite, for example, opening the file with sqlite3.connect("file:app.db?mode=ro", uri=True) makes every write fail with OperationalError: attempt to write a readonly database (tested on a temporary database). The same criterion applies to files: limit the readable folder and the size read, and do not trust a path chosen by the model. Other measures are in Security and Safety: Jailbreaking and Guardrails.

The loop as a graph: LangGraph

The LangGraph documentation describes it as a low-level orchestration framework and runtime for long-running stateful agents: you define flows as graphs of shared state, nodes and edges, and you can mix deterministic steps and model-driven steps. In the listing the state is MessagesState (a list of messages), the agent node calls the model, the tools node executes the requested calls through ToolNode, and the conditional edge tools_condition sends the flow to tools if the last message contains tool calls, otherwise to the end. The edge from tools to agent closes the loop: it is the orchestrator from the first article, made explicit.

from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from fake_model import FakeModel

@tool
def find_order(id: str) -> str:
    """Returns the status of an order given its code (format A-NNNN)."""
    return {"A-1042": "out for delivery"}.get(id, "order does not exist")

tools = [find_order]
llm_with_tools = FakeModel().bind_tools(tools)

def agent(state: MessagesState):
    """Node that calls the model and adds its reply to the state."""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)   # tool_calls -> "tools", otherwise END
builder.add_edge("tools", "agent")
graph = builder.compile()

result = graph.invoke({"messages": [("user", "Where is my order A-1042?")]})
for m in result["messages"]:
    content = m.content or f"tool_calls={[(c['name'], c['args']) for c in m.tool_calls]}"
    print(f"{m.type:>5}: {content}")

# Output:
# human: Where is my order A-1042?
#    ai: tool_calls=[('find_order', {'id': 'A-1042'})]
#  tool: out for delivery
#    ai: Final answer: out for delivery

Run twice, with identical output. The printout shows the four messages of the exchange: the question, the call to find_order, the result "out for delivery" and the answer. If you remove the tools edge toward agent, the graph stops after the tool is executed and the final answer never arrives: it is the most common mistake of those who write the graph by hand. The advantage over a while loop is that every node is an observable and savable unit; the cost is more structure to write.

The shortcut: create_agent

If you do not need to change the shape of the loop, create_agent builds the same graph for you. The documentation describes it as a minimal, configurable wrapper around the model loop, to which you compose model, tools, prompt and middleware. The block repeats the example with the shortcut and confirms that the sequence of messages is the same.

from langchain.agents import create_agent
from langchain_core.tools import tool
from fake_model import FakeModel

@tool
def find_order(id: str) -> str:
    """Returns the status of an order given its code (format A-NNNN)."""
    return {"A-1042": "out for delivery"}.get(id, "order does not exist")

agent = create_agent(FakeModel(), [find_order],
                     system_prompt="You are an assistant for order status.")
out = agent.invoke({"messages": [("user", "Where is A-1042?")]})
print([m.type for m in out["messages"]])
print(out["messages"][-1].content)

# Output:
# ['human', 'ai', 'tool', 'ai']
# Final answer: out for delivery

The sequence human, ai, tool, ai is that of the hand-written graph. To decide between the two routes: start from create_agent and drop down to StateGraph only when you need nodes of your own, different paths depending on the state, or more agents in the same graph. Moving from one to the other is not a rethink: the graph underneath is the same.

Saved state: threads and checkpoints

By compiling the graph with a checkpointer (here InMemorySaver, which can also be imported as MemorySaver) the state is saved and resumed for each thread_id. According to the documentation, thread_id is the pointer with which the checkpointer knows which state to load. The block uses a fake node that declares how many messages it sees, so you can see persistence without a model: two requests on the same thread and one on a different thread.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AIMessage

def count(state: MessagesState):
    """Fake node: says how many messages it has seen, to show that the state persists."""
    return {"messages": [AIMessage(content=f"Messages in state: {len(state['messages']) + 1}")]}

builder = StateGraph(MessagesState)
builder.add_node("count", count)
builder.add_edge(START, "count")
builder.add_edge("count", END)
graph = builder.compile(checkpointer=InMemorySaver())

for thread, text in (("user-1", "My name is Federico"), ("user-1", "What is my name?"),
                     ("user-2", "Hello")):
    config = {"configurable": {"thread_id": thread}}
    out = graph.invoke({"messages": [("user", text)]}, config)
    print(thread, "->", out["messages"][-1].content)

# Output:
# user-1 -> Messages in state: 2
# user-1 -> Messages in state: 4
# user-2 -> Messages in state: 2

On the same thread the count goes from 2 to 4, because the second request starts from the messages of the first; on the thread user-2 it starts again from 2. In-memory saving disappears at every restart of the process: for real memory across sessions you need a durable store, and for long-term memory a separate design, the topic of Memory Systems: Context, History and Knowledge Graphs. Another risk: if you use a single thread_id for all users, you share everyone's conversation.

Human approval with interrupt

For irreversible actions (sending an email, moving money) the graph must be able to stop and wait for a person. In LangGraph this is done by interrupt() inside a node: it suspends execution, hands the caller the value passed, and resumes when it receives Command(resume=...) with the same thread_id. The checkpointer is required, because the state must survive the pause. In the listing the node asks for the approval of an email; the program prints the request, the pending state, then resumes with "approved".

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command, interrupt

def ask_approval(state: MessagesState):
    """Stops and hands the caller the description of the action to approve."""
    decision = interrupt({"action": "send_email", "to": "customer@example.com"})
    outcome = "sent" if decision == "approved" else "cancelled"
    return {"messages": [("assistant", f"Email {outcome}.")]}

builder = StateGraph(MessagesState)
builder.add_node("approve", ask_approval)
builder.add_edge(START, "approve")
builder.add_edge("approve", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "t1"}}
first = graph.invoke({"messages": [("user", "Notify the customer")]}, config)
print("interrupted:", first["__interrupt__"][0].value)
print("next node to run:", graph.get_state(config).next)
final = graph.invoke(Command(resume="approved"), config)
print("after resume:", final["messages"][-1].content)

# Output:
# interrupted: {'action': 'send_email', 'to': 'customer@example.com'}
# next node to run: ('approve',)
# after resume: Email sent.

In the output, the first call does not end the graph: it returns the payload under the __interrupt__ key, and get_state(config).next says that the next node to run is still approve. After the resume the function restarts from the beginning of the node (not from the line of the interrupt), which is why whatever precedes interrupt() in the node must have no side effects: it would repeat the effect at every resumption. Here the return value only decides the text, it sends nothing.

Limits and resilience

An agent in production has two ways of failing: it runs too long or it falls over because of an external error. Against the first, LangGraph has a step limit, recursion_limit, passed in the configuration. The block builds a stubborn node that always calls the same tool again and stops it at 6 steps, catching GraphRecursionError.

from langchain_core.messages import AIMessage
from langgraph.errors import GraphRecursionError
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool

@tool
def find_order(id: str) -> str:
    """Returns the status of an order."""
    return "order does not exist"

def stubborn(state: MessagesState):
    """Fake node: always calls the same tool again, never answering."""
    call = {"name": "find_order", "args": {"id": "A-9999"}, "id": f"c{len(state['messages'])}"}
    return {"messages": [AIMessage(content="", tool_calls=[call])]}

builder = StateGraph(MessagesState)
builder.add_node("agent", stubborn)
builder.add_node("tools", ToolNode([find_order]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()

try:
    graph.invoke({"messages": [("user", "Where is A-9999?")]}, {"recursion_limit": 6})
except GraphRecursionError as e:
    print(type(e).__name__)
    print(str(e).splitlines()[0])

# Output:
# GraphRecursionError
# Recursion limit of 6 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.

The exception message says that the limit was reached without a stop condition and how to raise it. Raising it is never the first answer: if the agent is spinning idle, a higher limit spends more calls for the same result. Usually you catch the exception, return to the user a partial answer or an escalation, and log the trace to understand why the stop condition did not fire.

Against external errors, every Runnable (models included) has with_retry and with_fallbacks. The block uses a model that fails with ConnectionError a set number of times, to show three cases: the retry that succeeds on the third call, the fallback to a second model, and the retry that gives up and lets the error rise.

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult

class Flaky(BaseChatModel):
    """Fails with ConnectionError for the first `failures` calls, then answers."""
    failures: int = 2
    name: str = "A"
    calls: list = []

    @property
    def _llm_type(self): return "flaky"

    def _generate(self, messages, stop=None, run_manager=None, **kw):
        self.calls.append(1)
        if len(self.calls) <= self.failures:
            raise ConnectionError(f"{self.name}: network down (attempt {len(self.calls)})")
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content=f"answer from {self.name}"))])

# retry: three attempts in all, the third succeeds
a = Flaky(failures=2, name="A", calls=[])
r = a.with_retry(stop_after_attempt=3, wait_exponential_jitter=False,
                 retry_if_exception_type=(ConnectionError,))
print("retry   :", r.invoke("hello").content, "| calls:", len(a.calls))

# fallback: the first model always fails, the second answers
p = Flaky(failures=99, name="P", calls=[])
s = Flaky(failures=0, name="S", calls=[])
print("fallback:", p.with_fallbacks([s]).invoke("hello").content)

# retry is not enough: after the attempts the error propagates
c = Flaky(failures=99, name="C", calls=[])
try:
    c.with_retry(stop_after_attempt=2, wait_exponential_jitter=False,
                 retry_if_exception_type=(ConnectionError,)).invoke("hello")
except ConnectionError as e:
    print("error   :", e, "| calls:", len(c.calls))

# Output:
# retry   : answer from A | calls: 3
# fallback: answer from S
# error   : C: network down (attempt 2) | calls: 2

The three cases can be read in the output: with three attempts allowed and two failures the model answers on the third call; with the fallback the second model answers because the first always fails; with two attempts and a permanent failure the error rises after two calls. The retry_if_exception_type parameter limits the attempts to transient errors, and it must be chosen with care: retrying a validation error repeats the same error at your expense. On real models add a total time cap, because retry and fallback add up the latencies. The time of a call to a real model was not measured here.

LCEL: composing without an agent

Not every problem is an agent. LCEL (LangChain Expression Language) composes Runnables with the | operator: prompt, model, parser. RunnableParallel runs independent branches on the same input, RunnableBranch picks the first chain whose condition is true and uses the last as the default. It is the right level for the workflows of the first article, where the steps are fixed and the model fills in only the fields. The block uses FakeListChatModel for the sequence and simple functions in place of models in the other two cases.

from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnableParallel

llm = FakeListChatModel(responses=["A state graph for agents."])   # fake model
parser = StrOutputParser()

# Sequence: prompt | model | parser
chain = ChatPromptTemplate.from_messages([("human", "{question}")]) | llm | parser
print("sequence:", chain.invoke({"question": "What is LangGraph?"}))

# Parallel: three branches on the same input (here plain functions instead of models)
parallel = RunnableParallel(
    words=RunnableLambda(lambda x: len(x["text"].split())),
    upper=RunnableLambda(lambda x: x["text"].upper()),
    language=RunnableLambda(lambda x: x["language"]),
)
print("parallel:", parallel.invoke({"text": "agents with graphs", "language": "it"}))

# Conditional branch: the first true condition picks the chain; the last is the default
branch = RunnableBranch(
    (lambda x: x["language"] == "it", RunnableLambda(lambda x: "Italian branch")),
    (lambda x: x["language"] == "en", RunnableLambda(lambda x: "English branch")),
    RunnableLambda(lambda x: "default branch"),
)
print("branch  :", [branch.invoke({"language": l}) for l in ("it", "en", "fr")])

# Output:
# sequence: A state graph for agents.
# parallel: {'words': 3, 'upper': 'AGENTS WITH GRAPHS', 'language': 'it'}
# branch  : ['Italian branch', 'English branch', 'default branch']

The sequence returns the fake model's preset answer; the parallel produces a dictionary with one key per branch; the branch chooses based on the language and falls to the default for fr. The order of the conditions matters: the first true one wins. If a task is solved this way, an agent is an extra cost, and the rule of the first article (Introduction, What They Are and When to Use Them) remains valid.

Frequently asked questions

What you can do now

You can recognize an outdated tutorial from its import, choose between create_agent and StateGraph, define tools with a schema the model can read, save state per thread, stop a graph for human approval and impose limits and fallbacks. You can also test the wiring with a fake model before spending a single call. The loop that the framework automates is the one written by hand in Fundamentals: OODA Loop, ReAct and Tool Calling; a constructed example of a research assistant on a graph can be found in Case Study: Autonomous Research Assistant, and architectures with multiple roles in Multi-Agent Orchestration. In the next article, CrewAI: Collaborative Agent Teams, we change paradigm: not a graph of steps, but a team of roles.