LangChain has max_iterations. That’s not a budget limit — here’s how to add one.

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(
    model="gpt-4o",
    max_tokens=2000,      # bound completion cost per call
    temperature=0,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Use the tools provided."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools=TOOLS, prompt=prompt)

executor = AgentExecutor(
    agent=agent,
    tools=TOOLS,
    max_iterations=10,         # bound number of steps
    max_execution_time=30,     # bound wall-clock time (seconds)
    handle_parsing_errors=True,
    verbose=False,
)

result = executor.invoke({"input": user_task})

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
from typing import Any

class BudgetCallbackHandler(BaseCallbackHandler):
    """Raises BudgetExceededError if the run exceeds max_usd."""

    PRICING = {
        "gpt-4o":      (2.50, 10.00),   # (input, output) per 1M tokens
        "gpt-4o-mini": (0.15,  0.60),
        "claude-sonnet-4-6": (3.00, 15.00),
    }

    def __init__(self, max_usd: float, model: str = "gpt-4o"):
        super().__init__()
        self.max_usd = max_usd
        self.model = model
        self.spent = 0.0
        self.input_px, self.output_px = self.PRICING.get(model, (2.50, 10.00))

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        for gen_list in response.generations:
            for gen in gen_list:
                usage = getattr(gen.message, "usage_metadata", None)
                if usage:
                    call_cost = (
                        usage.get("input_tokens", 0) * self.input_px +
                        usage.get("output_tokens", 0) * self.output_px
                    ) / 1_000_000
                    self.spent += call_cost
        if self.spent > self.max_usd:
            raise RuntimeError(
                f"Budget exceeded: ${self.spent:.4f} spent (cap ${self.max_usd:.2f})"
            )

# Usage
budget_handler = BudgetCallbackHandler(max_usd=1.50, model="gpt-4o")

result = executor.invoke(
    {"input": user_task},
    config={"callbacks": [budget_handler]},
)
# handler.spent contains total cost after the run

from langchain_openai import ChatOpenAI
from langchain.schema import BaseMessage
from runguard import guard, BudgetExceededError, LoopDetectedError

# Subclass ChatOpenAI to add RunGuard
class GuardedChatOpenAI(ChatOpenAI):
    """ChatOpenAI subclass with per-run budget cap and loop detection."""

    def __init__(self, *args, max_usd: float = 2.0, **kwargs):
        super().__init__(*args, **kwargs)
        object.__setattr__(self, "_max_usd", max_usd)
        object.__setattr__(self, "_run_guard", None)

    def _make_guard(self):
        def _inner(messages, **kw):
            response = super(GuardedChatOpenAI, self)._generate(messages, **kw)
            # Extract cost from usage metadata
            usage = getattr(response.generations[0][0].message, "usage_metadata", {})
            usd = (
                usage.get("input_tokens", 0) * 2.50 +
                usage.get("output_tokens", 0) * 10.0
            ) / 1_000_000
            # Extract tool-call signature
            msg = response.generations[0][0].message
            sig = "end_turn"
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                sig = msg.tool_calls[0]["name"]
            return {"response": response, "usd": usd, "sig": sig}
        return guard(
            _inner,
            budget={"max_usd": self._max_usd},
            loop={"repeats": 3, "max_cycle_len": 8},
        )

    def _generate(self, messages: list[BaseMessage], **kwargs):
        if self._run_guard is None:
            object.__setattr__(self, "_run_guard", self._make_guard())
        result = self._run_guard(messages, **kwargs)
        return result["response"]

# Use it in AgentExecutor or LangGraph exactly like ChatOpenAI
llm = GuardedChatOpenAI(
    model="gpt-4o",
    max_usd=2.00,   # $2 per-run hard cap
    temperature=0,
)

# Or for LangGraph
from langgraph.prebuilt import create_react_agent
agent_graph = create_react_agent(llm, tools=TOOLS)

try:
    result = agent_graph.invoke({"messages": [("user", user_task)]})
except BudgetExceededError as e:
    print(f"Budget: ${e.spent:.3f} of $2.00")
except LoopDetectedError as e:
    print(f"Loop: {e.pattern!r} × {e.repeats}")