Put Limits Around a LangChain Agent

The manual reader works for pump P7. A technician then asks about P8, whose manual has not been uploaded. Botie tries another lookup, receives the same missing result, and tries again. The loop needs a stopping rule that does not depend on the model deciding it has done enough.

LangChain provides an agent runtime that coordinates messages and tools. We will keep the task small, give it a read-only tool, and put a call limit around execution. The workshop lets you build the permission and budget decision yourself.

Combine the application’s permitted tool list with the remaining call budget. The output names the reason a request stops.

execute read_manual

A message describes what the technician asked. The model may return answer text or structured requests to call tools. The runtime executes permitted tools and adds their results to the conversation so the model can continue.

create_agent constructs this repeated model-and-tool process. Its interface is useful when that loop fits the task. If every step is predetermined, a small ordinary function may be clearer. Reaching for an agent is a design choice, not a requirement for using a language model.

We are using the same maintenance question as before so the runtime change does not hide a change in the task.

A model response can contain a tool-call ID as well as a tool name and arguments. The result must refer to the matching call. That pairing lets the next model request distinguish one lookup from another, including when several tools were requested together.

Do not reduce the whole conversation to the last sentence while debugging. Inspect which arguments were requested and which result came back. An empty lookup and a network failure need different handling even if both prevented a useful answer.

LangChain returns agent state containing messages. The final message alone can hide the reason a run became expensive.

requested = {"id": "call-7", "name": "read_manual", "args": {"asset": "P7"}}
results = [
    {"call_id": "call-8", "found": False},
    {"call_id": "call-7", "found": True},
]
matched = next(item for item in results if item["call_id"] == requested["id"])
print(requested["args"]["asset"])
print(matched["found"])

P7
True

The @tool decorator uses a Python function as a tool definition. Type annotations describe the arguments, while the docstring tells the model what the operation does. Our example accepts an asset ID and returns either an excerpt or an explicit missing result.

For a real service, scope the lookup to the authenticated technician’s permitted equipment before reading data. The model-supplied asset string is a request, not evidence that the user owns that asset. A service layer can enforce that check regardless of which framework invokes it.

The example’s one-entry dictionary is deliberately invented. It demonstrates lookup semantics without giving the agent access to real maintenance records.

Middleware is code that runs at defined points around the agent loop. Hooks can inspect model requests or wrap tool execution. This gives the application a place to enforce limits and record what happened without burying every check in prompt text.

A limit on model calls and a limit on tool calls measure different things. One model response may request several tools. Choose the quantity that bounds the work you care about, and make the stopping behaviour visible to the caller.

The workshop checks an individual tool request against a count of executed calls. It is a small policy exercise, not a reproduction of every framework middleware counter.

This server recipe uses create_agent with the read_manual tool and separate model and tool limits. The selected error behaviour raises when a limit is exceeded. In a web service, catch that failure at the application boundary and return an unsuccessful result the interface can explain.

The tool only reads from a fixture. There is no create_job tool to call. Keep that narrow surface while learning how the messages and limits behave, then add capabilities with their own tests.

Provider responses vary. Compare the actual call sequence and final evidence with the task requirements instead of copying a displayed answer.

import os
from langchain.agents import create_agent
from langchain.tools import tool
from langchain.agents.middleware import (
    ModelCallLimitMiddleware, ToolCallLimitMiddleware,
)

@tool
def read_manual(asset: str) -> dict:
    """Read the invented inspection excerpt for a pump asset."""
    manuals = {"P7": {"id": "manual-7", "text": "Inspect the mounting."}}
    return manuals.get(asset, {"error": "No excerpt for this asset"})

agent = create_agent(
    model=os.environ["PERCEPTRON_LAB_AGENT_MODEL"],
    tools=[read_manual],
    system_prompt="Answer from the manual and cite its ID. "
                  "If no excerpt exists, say so. Do not schedule work.",
    middleware=[
        ModelCallLimitMiddleware(run_limit=4, exit_behavior="error"),
        ToolCallLimitMiddleware(run_limit=2, exit_behavior="error"),
    ],
)
state = agent.invoke({"messages": [
    {"role": "user", "content": "What should I inspect on P7?"}
]})
print(state["messages"][-1].content)

A temporary connection failure may justify another attempt. An asset absent from the catalogue generally needs different evidence, not immediate repetition of the same lookup. Classify these outcomes before choosing retries.

When an operation changes external state, a timeout can mean the effect happened but the reply was lost. Retrying create_job blindly can create a duplicate. Keep write operations separate from our read example and give them an idempotency key that the receiving service enforces.

A bounded retry can still repeat a mistake. Investigate whether another attempt can supply information the previous attempt lacked.

Our offline gate receives the allowed tool names and used-call count from trusted application state. It checks permission first, then whether another execution fits. These inputs must not come from the model’s claim that it has permission or from its own reported count.

For concurrent requests, reserving budget needs an atomic operation in shared state. A pair of workers can both read one remaining slot unless the runtime coordinates them. The worksheet is sequential so that you can inspect the underlying comparison before adding concurrency.

Allowed = ["read_manual"]
Used calls = 2
Limit = 2
Requested = "read_manual"
Next route = "budget exhausted"

The limit is reached when used calls equals the limit. There is no spare call at equality.

Keep a small set of maintenance fixtures while trying implementations. Does each version preserve citation IDs, reject disallowed operations and stop when evidence is missing? Can you inspect a failure without reading framework internals?

Pydantic AI and LangChain both support more than the features introduced here. Compare the behaviour your service needs and the code your team can maintain. Moving to a larger abstraction does not remove the need for the contracts we have already built.

The next lesson takes explicit control of state transitions when a single model-tool loop no longer describes the work clearly.