mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-22 15:42:25 +02:00
## Summary
- Filter `ToolMessage` from v3 `run.messages` streaming (handler +
`MessagesTransformer`) so tool results do not appear as chat text
deltas.
- Normalize `ToolCallStream.output` in `ToolCallTransformer` so live and
serialized `ToolMessage` payloads resolve to raw `content`.
- Add regression tests for message filtering and tool-output unwrapping.
<details>
<summary>Reproducible script</summary>
```python
"""Repro: v3 tool results must not leak through run.messages."""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Sequence
from typing import Any
from uuid import uuid4
from deepagents import create_deep_agent
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import BaseChatModel, LanguageModelInput
from langchain_core.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.runnables import Runnable
from langchain_core.tools import BaseTool, tool
from pydantic import Field
TOOL_RESULT_SENTINEL = "[]"
class ScriptedChatModel(BaseChatModel):
responses: list[AIMessage] = Field(default_factory=list)
tools: Sequence[dict[str, Any] | type | Callable | BaseTool] = ()
_idx: int = 0
@property
def _llm_type(self) -> str:
return "scripted"
def _generate(
self,
messages: Sequence[Any],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs: Any,
) -> ChatResult:
del messages, stop, run_manager, kwargs
idx = min(self._idx, len(self.responses) - 1)
self._idx += 1
return ChatResult(generations=[ChatGeneration(message=self.responses[idx])])
def bind_tools(
self,
tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
*,
tool_choice: str | None = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]:
del tool_choice, kwargs
self.tools = tools
return self
@tool
def list_items() -> str:
"""List available items."""
return TOOL_RESULT_SENTINEL
def _tool_call_message() -> AIMessage:
return AIMessage(
content="",
tool_calls=[
ToolCall(id="call_list", name="list_items", args={}),
],
)
def _extract_text_delta(event: Any) -> str | None:
if isinstance(event, dict):
if event.get("event") != "content-block-delta":
return None
delta = event.get("delta")
if isinstance(delta, dict) and delta.get("type") == "text-delta":
text = delta.get("text")
return text if isinstance(text, str) else None
elif getattr(event, "event", None) == "content-block-delta":
delta = getattr(event, "delta", None)
if isinstance(delta, dict) and delta.get("type") == "text-delta":
text = delta.get("text")
return text if isinstance(text, str) else None
text = getattr(delta, "text", None)
return text if isinstance(text, str) else None
return None
async def _tool_output(tool_call: Any) -> Any:
output = getattr(tool_call, "output", None)
if callable(output):
return await output()
if hasattr(output, "__await__"):
return await output
return output
async def main() -> None:
model = ScriptedChatModel(
responses=[
_tool_call_message(),
AIMessage(content="No items found."),
]
)
agent = create_deep_agent(model=model, tools=[list_items])
run = await agent.astream_events(
{"messages": [HumanMessage(content="List items")]},
version="v3",
configurable={"thread_id": str(uuid4())},
recursion_limit=50,
)
async def collect_message_texts() -> list[str]:
texts: list[str] = []
async for message_stream in run.messages:
async for event in message_stream:
text = _extract_text_delta(event)
if text is not None:
texts.append(text)
return texts
async def collect_tool_outputs() -> list[Any]:
outputs: list[Any] = []
async for tool_call in run.tool_calls:
outputs.append(await _tool_output(tool_call))
return outputs
message_texts, tool_outputs, final_state = await asyncio.gather(
collect_message_texts(),
collect_tool_outputs(),
run.output(),
)
final_messages = final_state["messages"]
tool_message = next((m for m in final_messages if isinstance(m, ToolMessage)), None)
print("run.messages text deltas:", message_texts)
print("run.tool_calls outputs:", tool_outputs)
print("final state message roles:", [m.type for m in final_messages])
if TOOL_RESULT_SENTINEL in message_texts:
raise AssertionError("Tool result leaked through run.messages.")
if TOOL_RESULT_SENTINEL not in tool_outputs:
raise AssertionError("Tool output was not surfaced through run.tool_calls.")
if tool_message is None or tool_message.tool_call_id != "call_list":
raise AssertionError("Final state does not contain the expected ToolMessage.")
print("Reproduction passed: tool output stayed out of run.messages.")
if __name__ == "__main__":
asyncio.run(main())
```
</details>
<details>
<summary>Current behavior</summary>
```text
run.messages text deltas: ['[]', 'No items found.']
run.tool_calls outputs: [ToolMessage(content='[]', ...)]
final state message roles: ['human', 'ai', 'tool', 'ai']
AssertionError: Tool result leaked through run.messages.
```
</details>
<details>
<summary>Expected behavior</summary>
```text
run.messages text deltas: ['No items found.']
run.tool_calls outputs: ['[]']
final state message roles: ['human', 'ai', 'tool', 'ai']
Reproduction passed: tool output stayed out of run.messages.
```
</details>
## Test plan
- [ ] `uv run --project libs/langgraph pytest
libs/langgraph/tests/test_stream_messages_transformer.py`
- [ ] `uv run --project libs/prebuilt pytest
libs/prebuilt/tests/test_tool_call_transformer.py`
- [ ] `uv run --project libs/langgraph ruff check` (touched files)
- [ ] `uv run --project libs/prebuilt ruff check` (touched files)
Related:
[langchain-ai/langchainjs#10900](https://github.com/langchain-ai/langchainjs/pull/10900)
LangGraph Prebuilt
This library defines high-level APIs for creating and executing LangGraph agents and tools.
Important
This library is meant to be bundled with
langgraph, don't install it directly
Agents
langgraph-prebuilt provides an implementation of a tool-calling ReAct-style agent - create_react_agent:
pip install langchain-anthropic
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
# Define the tools for the agent to use
def search(query: str):
"""Call to surf the web."""
# This is a placeholder, but don't tell the LLM that...
if "sf" in query.lower() or "san francisco" in query.lower():
return "It's 60 degrees and foggy."
return "It's 90 degrees and sunny."
tools = [search]
model = ChatAnthropic(model="claude-3-7-sonnet-latest")
app = create_react_agent(model, tools)
# run the agent
app.invoke(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
)
Tools
ToolNode
langgraph-prebuilt provides an implementation of a node that executes tool calls - ToolNode:
from langgraph.prebuilt import ToolNode
from langchain_core.messages import AIMessage
def search(query: str):
"""Call to surf the web."""
# This is a placeholder, but don't tell the LLM that...
if "sf" in query.lower() or "san francisco" in query.lower():
return "It's 60 degrees and foggy."
return "It's 90 degrees and sunny."
tool_node = ToolNode([search])
tool_calls = [{"name": "search", "args": {"query": "what is the weather in sf"}, "id": "1"}]
ai_message = AIMessage(content="", tool_calls=tool_calls)
# execute tool call
tool_node.invoke({"messages": [ai_message]})
ValidationNode
langgraph-prebuilt provides an implementation of a node that validates tool calls against a pydantic schema - ValidationNode:
from pydantic import BaseModel, field_validator
from langgraph.prebuilt import ValidationNode
from langchain_core.messages import AIMessage
class SelectNumber(BaseModel):
a: int
@field_validator("a")
def a_must_be_meaningful(cls, v):
if v != 37:
raise ValueError("Only 37 is allowed")
return v
validation_node = ValidationNode([SelectNumber])
validation_node.invoke({
"messages": [AIMessage("", tool_calls=[{"name": "SelectNumber", "args": {"a": 42}, "id": "1"}])]
})
Agent Inbox
The library contains schemas for using the Agent Inbox with LangGraph agents. Learn more about how to use Agent Inbox here.
from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse
def my_graph_function():
# Extract the last tool call from the `messages` field in the state
tool_call = state["messages"][-1].tool_calls[0]
# Create an interrupt
request: HumanInterrupt = {
"action_request": {
"action": tool_call['name'],
"args": tool_call['args']
},
"config": {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": False,
"allow_accept": False
},
"description": _generate_email_markdown(state) # Generate a detailed markdown description.
}
# Send the interrupt request inside a list, and extract the first response
response = interrupt([request])[0]
if response['type'] == "response":
# Do something with the response
...