mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-17 21:25:46 +02:00
Christian BromannandGitHub
ea44df3476
## 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)