This is tool emulation middleware. The idea is to help test out an agent
that may have some tools that either take a long time to run or are
expensive to set up. This could allow simulating the behavior a bit.
Easiest to review side by side (not inline)
* Adding `dict` type requests + responses so that we can ship config w/
interrupts. Also more extensible.
* Keeping things generic in terms of `interrupt_on` rather than
`tool_config`
* Renaming allowed decisions -- approve, edit, reject
* Draws differentiation between actions (requested + performed by the
agent), in this case tool calls, though we generalize beyond that and
decisions - human feedback for said actions
New request structure
```py
class Action(TypedDict):
"""Represents an action with a name and arguments."""
name: str
"""The type or name of action being requested (e.g., "add_numbers")."""
arguments: dict[str, Any]
"""Key-value pairs of arguments needed for the action (e.g., {"a": 1, "b": 2})."""
DecisionType = Literal["approve", "edit", "reject"]
class ReviewConfig(TypedDict):
"""Policy for reviewing a HITL request."""
action_name: str
"""Name of the action associated with this review configuration."""
allowed_decisions: list[DecisionType]
"""The decisions that are allowed for this request."""
description: NotRequired[str]
"""The description of the action to be reviewed."""
arguments_schema: NotRequired[dict[str, Any]]
"""JSON schema for the arguments associated with the action, if edits are allowed."""
class HITLRequest(TypedDict):
"""Request for human feedback on a sequence of actions requested by a model."""
action_requests: list[Action]
"""A list of agent actions for human review."""
review_configs: list[ReviewConfig]
"""Review configuration for all possible actions."""
```
New response structure
```py
class ApproveDecision(TypedDict):
"""Response when a human approves the action."""
type: Literal["approve"]
"""The type of response when a human approves the action."""
class EditDecision(TypedDict):
"""Response when a human edits the action."""
type: Literal["edit"]
"""The type of response when a human edits the action."""
edited_action: Action
"""Edited action for the agent to perform.
Ex: for a tool call, a human reviewer can edit the tool name and args.
"""
class RejectDecision(TypedDict):
"""Response when a human rejects the action."""
type: Literal["reject"]
"""The type of response when a human rejects the action."""
message: NotRequired[str]
"""The message sent to the model explaining why the action was rejected."""
Decision = ApproveDecision | EditDecision | RejectDecision
class HITLResponse(TypedDict):
"""Response payload for a HITLRequest."""
decisions: list[Decision]
"""The decisions made by the human."""
```
User facing API:
NEW
```py
HumanInTheLoopMiddleware(interrupt_on={
'send_email': True,
# can also use a callable for description that takes tool call, state, and runtime
'execute_sql': {
'allowed_decisions': ['approve', 'edit', 'reject'],
'description': 'please review sensitive tool execution'},
}
})
Command(resume={"decisions": [{"type": "approve"}, {"type": "reject": "message": "db down"}]})
```
OLD
```py
HumanInTheLoopMiddleware(interrupt_on={
'send_email': True,
'execute_sql': {
'allow_accept': True,
'allow_edit': True,
'allow_respond': True,
description='please review sensitive tool execution'
},
})
Command(resume=[{"type": "approve"}, {"type": "reject": "message": "db down"}])
```
Refactor tool call middleware from generator-based to handler-based
pattern
Simplifies on_tool_call middleware by replacing the complex generator
protocol with a straightforward handler pattern. Instead of yielding
requests and receiving results via .send(),
handlers now receive an execute callable that can be invoked multiple
times for retry logic.
Before vs. After
Before (Generator):
```python
class RetryMiddleware(AgentMiddleware):
def on_tool_call(self, request, state, runtime):
for attempt in range(3):
response = yield request # Yield request, receive result via .send()
if is_valid(response) or attempt == 2:
return # Final result is last value sent to generator
```
After (Handler):
```python
class RetryMiddleware(AgentMiddleware):
def on_tool_call(self, request, handler):
for attempt in range(3):
result = handler(request) # Direct function call
if is_valid(result):
return result
return result
```
Follow up after this PR:
* Rename the interceptor to wrap_tool_call
* Fix the async path for the ToolNode
1. Main fix: when we don't have a response format or middleware, don't
draw a conditional edge back to the loop entrypoint (self loop on model)
2. Supplementary fix: when we jump to `end` and there is an
`after_agent` hook, jump there instead of `__end__`
Other improvements -- I can remove these if they're more harmful than
helpful
1. Use keyword only arguments for edge generator functions for clarity
2. Rename args to `model_destination` and `end_destination` for clarity
This reduces confusion w/ types like `AgentState`, different arg names,
etc.
Second attempt, following
https://github.com/langchain-ai/langchain/pull/33249
* Ability to pass through `cache` and name in `create_agent` as
compilation args for the agent
* Right now, removing `test_react_agent.py` but we should add these
tests back as implemented w/ the new agent
* Add conditional edge when structured output tools are present to allow
for retries
* Rename `tracking` to `model_call_limit` to be consistent w/ tool call
limits
We need in the future (I'm happy to own):
* Significant test refactor
* Significant test overhaul where we emphasize and enforce coverage
- retry_model_request hook lets a middleware decide to retry a failed
model request, with full ability to modify as much or as little of the
request before doing so
- ModelFallbackMiddleware tries each fallback model in order, until one
is successful, or fallback list is exhausted
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
* Add llm based tool selection middleware.
* Note that we might want some form of caching for when the agent is
inside an active tool calling loop as the tool selection isn't expected
to change during that time.
API:
```python
class LLMToolSelectorMiddleware(AgentMiddleware):
"""Uses an LLM to select relevant tools before calling the main model.
When an agent has many tools available, this middleware filters them down
to only the most relevant ones for the user's query. This reduces token usage
and helps the main model focus on the right tools.
Examples:
Limit to 3 tools:
```python
from langchain.agents.middleware import LLMToolSelectorMiddleware
middleware = LLMToolSelectorMiddleware(max_tools=3)
agent = create_agent(
model="openai:gpt-4o",
tools=[tool1, tool2, tool3, tool4, tool5],
middleware=[middleware],
)
```
Use a smaller model for selection:
```python
middleware = LLMToolSelectorMiddleware(model="openai:gpt-4o-mini", max_tools=2)
```
"""
def __init__(
self,
*,
model: str | BaseChatModel | None = None,
system_prompt: str = DEFAULT_SYSTEM_PROMPT,
max_tools: int | None = None,
always_include: list[str] | None = None,
) -> None:
"""Initialize the tool selector.
Args:
model: Model to use for selection. If not provided, uses the agent's main model.
Can be a model identifier string or BaseChatModel instance.
system_prompt: Instructions for the selection model.
max_tools: Maximum number of tools to select. If the model selects more,
only the first max_tools will be used. No limit if not specified.
always_include: Tool names to always include regardless of selection.
These do not count against the max_tools limit.
"""
```
```python
"""Test script for LLM tool selection middleware."""
from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware
from langchain_core.tools import tool
@tool
def get_weather(location: str) -> str:
"""Get current weather for a location."""
return f"Weather in {location}: 72°F, sunny"
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Perform mathematical calculations."""
return f"Result of {expression}: 42"
@tool
def send_email(to: str, subject: str) -> str:
"""Send an email to someone."""
return f"Email sent to {to} with subject: {subject}"
@tool
def get_stock_price(symbol: str) -> str:
"""Get current stock price for a symbol."""
return f"Stock price for {symbol}: $150.25"
@tool
def translate_text(text: str, target_language: str) -> str:
"""Translate text to another language."""
return f"Translated '{text}' to {target_language}"
@tool
def set_reminder(task: str, time: str) -> str:
"""Set a reminder for a task."""
return f"Reminder set: {task} at {time}"
@tool
def get_news(topic: str) -> str:
"""Get latest news about a topic."""
return f"Latest news about {topic}"
@tool
def book_flight(destination: str, date: str) -> str:
"""Book a flight to a destination."""
return f"Flight booked to {destination} on {date}"
@tool
def get_restaurant_recommendations(city: str, cuisine: str) -> str:
"""Get restaurant recommendations."""
return f"Top {cuisine} restaurants in {city}"
# Create agent with tool selection middleware
middleware = LLMToolSelectorMiddleware(
model="openai:gpt-4o-mini",
max_tools=3,
)
agent = create_agent(
model="openai:gpt-4o",
tools=[
get_weather,
search_web,
calculate,
send_email,
get_stock_price,
translate_text,
set_reminder,
get_news,
book_flight,
get_restaurant_recommendations,
],
middleware=[middleware],
)
# Test with a query that should select specific tools
response = agent.invoke(
{"messages": [{"role": "user", "content": "I need to find restaurants"}]}
)
print(response)
```
- supports 6 well-known PII types (email, credit_card, ip, mac_address,
url)
- 4 handling strategies (block, redact, mask, hash)
- supports custom PII types with detector functions or regex
- the built-in types were chosen because they are common, and detection
can be reliably implemented with stdlib
Change response format strategy dynamically based on model.
After this PR there are two remaining issues:
- [ ] Review binding of tools used for output to ToolNode (shouldn't be
required)
- [ ] Update ModelRequest to also support the original schema provided
by the user (to correctly support auto mode)
Adding a `dynamic_prompt` decorator to support smoother devx for dynamic
system prompts
```py
from langchain.agents.middleware.types import dynamic_prompt, ModelRequest, AgentState
from langchain.agents.middleware_agent import create_agent
from langgraph.runtime import Runtime
from dataclasses import dataclass
from langchain_core.messages import HumanMessage
@dataclass
class Context:
user_name: str
@dynamic_prompt
def my_prompt(request: ModelRequest, state: AgentState, runtime: Runtime[Context]) -> str:
user_name = runtime.context.user_name
return (
f"You are a helpful assistant helping {user_name}. Please refer to the user as {user_name}."
)
agent = create_agent(model="openai:gpt-4o", middleware=[my_prompt]).compile()
result = agent.invoke({"messages": [HumanMessage("Hello")]}, context=Context(user_name="Sydney"))
for msg in result["messages"]:
msg.pretty_print()
"""
================================ Human Message =================================
Hello
================================== Ai Message ==================================
Hello Sydney! How can I assist you today?
"""
```
Need to decide - what information should we feed to this description
factory? Right now, feeding:
* state
* runtime
* tool call (so the developer doesn't have to search through the state's
messages for the corresponding tool call)
I can see a case for just passing tool call. But again, this abstraction
is semi-bound to interrupts for tools... though we pretend it's more
abstract than that.
Right now:
```py
def custom_description(state: AgentState, runtime: Runtime, tool_call: ToolCall) -> str:
"""Generate a custom description."""
return f"Custom: {tool_call['name']} with args {tool_call['args']}"
middleware = HumanInTheLoopMiddleware(
interrupt_on={
"tool_with_callable": {"allow_accept": True, "description": custom_description},
"tool_with_string": {"allow_accept": True, "description": "Static description"},
}
)
```
This PR adds a model call limit middleware that helps to manage:
* number of model calls during a run (helps w/ avoiding tool calling
loops) - implemented w/ `UntrackedValue`
* number of model calls on a thread (helps w/ avoiding lengthy convos) -
standard state
Concern here is w/ other middlewares overwriting the model call count...
we could use a `_` prefixed field?
The old `before_model_jump_to` classvar approach was quite clunky, this
is nicer imo and easier to document. Also moving from `jump_to` to
`can_jump_to` which is more idiomatic.
Before:
```py
class MyMiddleware(AgentMiddleware):
before_model_jump_to: ClassVar[list[JumpTo]] = ["end"]
def before_model(state, runtime) -> dict[str, Any]:
return {"jump_to": "end"}
```
After
```py
class MyMiddleware(AgentMiddleware):
@hook_config(can_jump_to=["end"])
def before_model(state, runtime) -> dict[str, Any]:
return {"jump_to": "end"}
```
This makes branching **much** more simple internally and helps greatly
w/ type safety for users. It just allows for one signature on hooks
instead of multiple.
Opened after https://github.com/langchain-ai/langchain/pull/33164
ballooned more than expected, w/ branching for:
* sync vs async
* runtime vs no runtime (this is self imposed)
**This also removes support for nodes w/o `runtime` in the signature.**
We can always go back and add support for nodes w/o `runtime`.
I think @christian-bromann's idea to re-export `runtime` from
langchain's agents might make sense due to the abundance of imports
here.
Check out the value of the change based on this diff:
https://github.com/langchain-ai/langchain/pull/33176
- **Description:** Changing the key from `response` to
`structured_response` for middleware agent to keep it sync with agent
without middleware. This a breaking change.
- **Issue:** #33154
Porting the [planning
middleware](39c0138d0f/src/deepagents/middleware.py (L21))
over from deepagents.
Also adding the ability to configure:
* System prompt
* Tool description
```py
from langchain.agents.middleware.planning import PlanningMiddleware
from langchain.agents import create_agent
agent = create_agent("openai:gpt-4o", middleware=[PlanningMiddleware()])
result = await agent.invoke({"messages": [HumanMessage("Help me refactor my codebase")]})
print(result["todos"]) # Array of todo items with status tracking
```
Multiple improvements to HITL flow:
* On a `response` type resume, we should still append the tool call to
the last AIMessage (otherwise we have a ToolResult without a
corresponding ToolCall)
* When all interrupts have `response` types (so there's no pending tool
calls), we should jump back to the first node (instead of end) as we
enforced in the previous `post_model_hook_router`
* Added comments to `model_to_tools` router so clarify all of the
potential exit conditions
Additionally:
* Lockfile update to use latest LG alpha release
* Added test for `jump_to` behaving ephemerally, this was fixed in LG
but surfaced as a bug w/ `jump_to`.
* Bump version to v1.0.0a10 to prep for alpha release
---------
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
