Skip to content

Commit 082bf3a

Browse files
Merge pull request #3202 from dgesteves/fix/langgraph-failed-tool-call-result
fix(langgraph): stream TOOL_CALL_RESULT for a tool call that fails
2 parents 73b04fd + d1b5dd8 commit 082bf3a

2 files changed

Lines changed: 407 additions & 100 deletions

File tree

‎integrations/langgraph/python/ag_ui_langgraph/agent.py‎

Lines changed: 131 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
from langchain_core.runnables import RunnableConfig, ensure_config
2020
from langchain_core.runnables.config import merge_configs
2121
from langchain_core.messages import AIMessage, HumanMessage
22+
from langgraph.errors import GraphBubbleUp
2223
from langgraph.types import Command
2324

2425
from .types import (
@@ -169,6 +170,35 @@ def _command_update_tool_messages(messages: Any) -> List[ToolMessage]:
169170
return tool_messages
170171

171172

173+
def _node_output_tool_messages(output: Any) -> List[ToolMessage]:
174+
"""Extract ToolMessages from node outputs, including Command updates."""
175+
if isinstance(output, Command):
176+
update = output.update
177+
return _command_update_tool_messages(
178+
update.get("messages") if isinstance(update, dict) else update
179+
)
180+
if isinstance(output, ToolMessage):
181+
return [output]
182+
if isinstance(output, dict):
183+
return _node_output_tool_messages(output.get("messages"))
184+
if isinstance(output, (list, tuple)):
185+
return [message for item in output for message in _node_output_tool_messages(item)]
186+
return []
187+
188+
189+
def _tool_invocation_key(tool_event: Any, tool_call_id: Any) -> Optional[tuple]:
190+
"""Key a tool call by the run that executed the tool and its raw tool_call_id.
191+
192+
The tool's parent run is the node (e.g. ToolNode) whose on_chain_end
193+
returns the call's result. A raw tool_call_id alone can repeat across
194+
turns and across concurrent graphs.
195+
"""
196+
parent_ids = (tool_event or {}).get("parent_ids") or []
197+
if not parent_ids or not isinstance(tool_call_id, str):
198+
return None
199+
return (parent_ids[-1], tool_call_id)
200+
201+
172202
# Lane key for root/supervisor-level streaming state (events with no subagent).
173203
# Subagent lanes use the derived subagent id (e.g. "tools:<uuid>").
174204
_ROOT_LANE = "__root__"
@@ -1644,6 +1674,10 @@ async def _handle_stream_events(self, input: RunAgentInput) -> AsyncGenerator[Pr
16441674
# tool message by (streamed in TOOL_CALL_RESULT, or echoed back in
16451675
# the input). See _present_streamed_tool_message_ids.
16461676
"tool_message_ids_by_call": {},
1677+
# (parent run id, raw tool_call_id) -> its on_tool_error event, for
1678+
# calls whose tool raised and that have no TOOL_CALL_RESULT yet. See
1679+
# _tool_invocation_key and OnChainEnd in _handle_single_event.
1680+
"failed_tool_calls": {},
16471681
"model_made_tool_call": False,
16481682
"state_reliable": True,
16491683
"active_subagents": {},
@@ -3711,60 +3745,9 @@ def _chunk_get(c: Any, key: str, default: Any = None) -> Any:
37113745
type(m).__name__,
37123746
)
37133747

3714-
# Process each tool message. Re-emit Start/Args/End only for
3715-
# tool_call_ids that did NOT already stream them through
3716-
# OnChatModelStream — checked per-id so nested tool execution
3717-
# (deepagents ``task`` -> subagent) doesn't cause the outer
3718-
# tool's Args to be emitted twice when the inner tool's
3719-
# OnToolEnd fires first.
37203748
for tool_msg in tool_messages:
3721-
# Same lane as the streamed call, so the resolve returns the
3722-
# public id its START carried (or the raw id when this call
3723-
# was never streamed and is being emitted here first).
3724-
public_call_id = self._resolve_public_tool_call_id(tool_msg.tool_call_id)
3725-
already_streamed = self._streamed_call_key(public_call_id) in self.active_run["streamed_tool_call_ids"]
3726-
if not already_streamed:
3727-
yield self._dispatch_event(
3728-
ToolCallStartEvent(
3729-
type=EventType.TOOL_CALL_START,
3730-
tool_call_id=public_call_id,
3731-
tool_call_name=tool_msg.name or event.get("name", ""),
3732-
parent_message_id=self._tool_call_owner(tool_msg.tool_call_id),
3733-
raw_event=event,
3734-
)
3735-
)
3736-
yield self._dispatch_event(
3737-
ToolCallArgsEvent(
3738-
type=EventType.TOOL_CALL_ARGS,
3739-
tool_call_id=public_call_id,
3740-
delta=dump_json_safe(event["data"].get("input", {})),
3741-
raw_event=event
3742-
)
3743-
)
3744-
yield self._dispatch_event(
3745-
ToolCallEndEvent(
3746-
type=EventType.TOOL_CALL_END,
3747-
tool_call_id=public_call_id,
3748-
raw_event=event
3749-
)
3750-
)
3751-
self.active_run["streamed_tool_call_ids"].discard(self._streamed_call_key(public_call_id))
3752-
3753-
# The tool's objects are never written to (it may reuse them),
3754-
# so an id-less message streams under its tool_call_id and
3755-
# the snapshot presents the stored copy under that id.
3756-
result_message_id = str(tool_msg.id or tool_msg.tool_call_id)
3757-
self._record_streamed_tool_message_id(event, tool_msg.tool_call_id, result_message_id)
3758-
yield self._dispatch_event(
3759-
ToolCallResultEvent(
3760-
type=EventType.TOOL_CALL_RESULT,
3761-
tool_call_id=public_call_id,
3762-
# Resolved through the same lane map as the START above.
3763-
message_id=self._resolve_public_message_id(result_message_id),
3764-
content=normalize_tool_content(tool_msg.content),
3765-
role="tool"
3766-
)
3767-
)
3749+
for result_event in self._tool_result_events(tool_msg, event):
3750+
yield result_event
37683751
self.active_run["model_made_tool_call"] = False
37693752
self.active_run["state_reliable"] = True
37703753
self.active_run["has_function_streaming"] = False
@@ -3781,53 +3764,8 @@ def _chunk_get(c: Any, key: str, default: Any = None) -> Any:
37813764
)
37823765
return
37833766

3784-
# Same lane as the streamed call, so the resolve returns the public
3785-
# id its START carried (or the raw id when this call was never
3786-
# streamed and is being emitted here first).
3787-
public_call_id = self._resolve_public_tool_call_id(tool_call_output.tool_call_id)
3788-
already_streamed = self._streamed_call_key(public_call_id) in self.active_run["streamed_tool_call_ids"]
3789-
if not already_streamed:
3790-
yield self._dispatch_event(
3791-
ToolCallStartEvent(
3792-
type=EventType.TOOL_CALL_START,
3793-
tool_call_id=public_call_id,
3794-
tool_call_name=tool_call_output.name or event.get("name", ""),
3795-
parent_message_id=self._tool_call_owner(tool_call_output.tool_call_id),
3796-
raw_event=event,
3797-
)
3798-
)
3799-
yield self._dispatch_event(
3800-
ToolCallArgsEvent(
3801-
type=EventType.TOOL_CALL_ARGS,
3802-
tool_call_id=public_call_id,
3803-
delta=dump_json_safe(event["data"]["input"]),
3804-
raw_event=event
3805-
)
3806-
)
3807-
yield self._dispatch_event(
3808-
ToolCallEndEvent(
3809-
type=EventType.TOOL_CALL_END,
3810-
tool_call_id=public_call_id,
3811-
raw_event=event
3812-
)
3813-
)
3814-
self.active_run["streamed_tool_call_ids"].discard(self._streamed_call_key(public_call_id))
3815-
3816-
# Same as the Command branch: add_messages stores an id-less
3817-
# message under a uuid of its own, and the snapshot presents it
3818-
# under the id streamed here.
3819-
result_message_id = str(tool_call_output.id or tool_call_output.tool_call_id)
3820-
self._record_streamed_tool_message_id(event, tool_call_output.tool_call_id, result_message_id)
3821-
yield self._dispatch_event(
3822-
ToolCallResultEvent(
3823-
type=EventType.TOOL_CALL_RESULT,
3824-
tool_call_id=public_call_id,
3825-
# Resolved through the same lane map as the START above.
3826-
message_id=self._resolve_public_message_id(result_message_id),
3827-
content=normalize_tool_content(tool_call_output.content),
3828-
role="tool"
3829-
)
3830-
)
3767+
for result_event in self._tool_result_events(tool_call_output, event):
3768+
yield result_event
38313769

38323770
self.active_run["model_made_tool_call"] = False
38333771
self.active_run["state_reliable"] = True
@@ -3843,6 +3781,99 @@ def _chunk_get(c: Any, key: str, default: Any = None) -> Any:
38433781
self.active_run["model_made_tool_call"] = False
38443782
self.active_run["state_reliable"] = True
38453783
self.active_run["has_function_streaming"] = False
3784+
# OnToolEnd never fires for this call. When ToolNode handles the
3785+
# error it returns an error ToolMessage instead, which OnChainEnd
3786+
# below streams as the call's result. A GraphBubbleUp (an
3787+
# interrupt, or a Command for the parent graph) is control flow,
3788+
# not a failure: ToolNode re-raises it and the call has no result,
3789+
# so it is not recorded.
3790+
data = event.get("data") or {}
3791+
key = _tool_invocation_key(event, data.get("tool_call_id"))
3792+
if key is not None and not isinstance(data.get("error"), GraphBubbleUp):
3793+
self.active_run.setdefault("failed_tool_calls", {})[key] = event
3794+
3795+
elif event_type == LangGraphEventTypes.OnChainEnd:
3796+
# Stream a failed call's error ToolMessage like a successful
3797+
# result, with the content ToolNode produced rather than the raw
3798+
# exception. A failure matches only the output of the run that
3799+
# executed its tool, so the root graph's final output, which
3800+
# repeats every tool message on the thread, never matches, and
3801+
# concurrent graphs that reuse a raw tool_call_id stay apart.
3802+
failed_tool_calls = self.active_run.get("failed_tool_calls")
3803+
if failed_tool_calls:
3804+
output = (event.get("data") or {}).get("output")
3805+
for tool_msg in _node_output_tool_messages(output):
3806+
key = (event.get("run_id"), tool_msg.tool_call_id)
3807+
error_event = failed_tool_calls.pop(key, None)
3808+
if error_event is not None:
3809+
for result_event in self._tool_result_events(tool_msg, error_event):
3810+
yield result_event
3811+
3812+
def _tool_result_events(self, tool_msg: ToolMessage, event: Any) -> Generator[ProcessedEvents, None, None]:
3813+
"""TOOL_CALL_RESULT for ``tool_msg``, announcing the call first if it never streamed.
3814+
3815+
``event`` is the tool's own on_tool_end or on_tool_error event: it
3816+
supplies the tool name and input for the announcement, and the
3817+
namespace the streamed message id is recorded under.
3818+
3819+
Start/Args/End are re-emitted only for tool_call_ids that did NOT
3820+
already stream them through OnChatModelStream. The check is per id so
3821+
nested tool execution (deepagents ``task`` -> subagent) doesn't cause
3822+
the outer tool's Args to be emitted twice when the inner tool's
3823+
OnToolEnd fires first.
3824+
"""
3825+
# Same lane as the streamed call, so the resolve returns the public id
3826+
# its START carried (or the raw id when this call was never streamed
3827+
# and is being emitted here first).
3828+
public_call_id = self._resolve_public_tool_call_id(tool_msg.tool_call_id)
3829+
already_streamed = self._streamed_call_key(public_call_id) in self.active_run["streamed_tool_call_ids"]
3830+
if not already_streamed:
3831+
yield self._dispatch_event(
3832+
ToolCallStartEvent(
3833+
type=EventType.TOOL_CALL_START,
3834+
tool_call_id=public_call_id,
3835+
tool_call_name=tool_msg.name or event.get("name", ""),
3836+
parent_message_id=self._tool_call_owner(tool_msg.tool_call_id),
3837+
raw_event=event,
3838+
)
3839+
)
3840+
yield self._dispatch_event(
3841+
ToolCallArgsEvent(
3842+
type=EventType.TOOL_CALL_ARGS,
3843+
tool_call_id=public_call_id,
3844+
delta=dump_json_safe(event["data"].get("input", {})),
3845+
raw_event=event
3846+
)
3847+
)
3848+
yield self._dispatch_event(
3849+
ToolCallEndEvent(
3850+
type=EventType.TOOL_CALL_END,
3851+
tool_call_id=public_call_id,
3852+
raw_event=event
3853+
)
3854+
)
3855+
self.active_run["streamed_tool_call_ids"].discard(self._streamed_call_key(public_call_id))
3856+
# The call has its result now. If an earlier attempt failed and a
3857+
# retry succeeded, the node's output must not stream it again.
3858+
(self.active_run.get("failed_tool_calls") or {}).pop(
3859+
_tool_invocation_key(event, tool_msg.tool_call_id), None)
3860+
3861+
# The tool's objects are never written to (it may reuse them), so an
3862+
# id-less message streams under its tool_call_id: add_messages stores
3863+
# it under a uuid of its own, and the snapshot presents the stored copy
3864+
# under the id streamed here.
3865+
result_message_id = str(tool_msg.id or tool_msg.tool_call_id)
3866+
self._record_streamed_tool_message_id(event, tool_msg.tool_call_id, result_message_id)
3867+
yield self._dispatch_event(
3868+
ToolCallResultEvent(
3869+
type=EventType.TOOL_CALL_RESULT,
3870+
tool_call_id=public_call_id,
3871+
# Resolved through the same lane map as the START above.
3872+
message_id=self._resolve_public_message_id(result_message_id),
3873+
content=normalize_tool_content(tool_msg.content),
3874+
role="tool"
3875+
)
3876+
)
38463877

38473878
def handle_reasoning_event(self, reasoning_data: LangGraphReasoning) -> Generator[ProcessedEvents, Any, None]:
38483879
# Invariant: reasoning events are dispatched from _handle_single_event,

0 commit comments

Comments
 (0)