From eae40934e237d151f3d7114d3b807dada0c1f44d Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Fri, 6 Nov 2026 10:49:42 -0500 Subject: [PATCH 1/2] fix(langchain): don't mark spans errored when a stream is closed early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit > 🤖 **This text was AI-generated.** When code stops reading a LangGraph stream partway through, Python closes the stream's generator with `GeneratorExit`. LangGraph passes that to `on_chain_error`, and `BraintrustCallbackHandler` logged it as `error=str(error)`. `str(GeneratorExit())` is `""`, so the root "LangGraph" span showed up as failed with an empty Error panel: ```python async for mode, chunk in graph.astream(..., config={"callbacks": [handler]}): if done(chunk): return # -> GeneratorExit -> on_chain_error -> span error: "" ``` The handler now treats `GeneratorExit` as an intentional stop, not a failure. A new `_error_for_span()` helper returns `None` for it and `str(error)` for everything else. All four error callbacks use it (`on_llm_error`, `on_chain_error`, `on_tool_error`, `on_retriever_error`), because LLM streams can be stopped early the same way: ``` GeneratorExit -> span ends, no error other errors -> span ends, error=str(error) (unchanged) ``` `asyncio.CancelledError` is still recorded as an error. Unlike the ADK integration, this handler does not treat it as cleanup, because it can also come from timeouts. Fixes SDK-797 --- .../integrations/langchain/callbacks.py | 15 ++++-- .../integrations/langchain/test_callbacks.py | 52 +++++++++++++++++-- 2 files changed, 58 insertions(+), 9 deletions(-) diff --git a/py/src/braintrust/integrations/langchain/callbacks.py b/py/src/braintrust/integrations/langchain/callbacks.py index dbd413b8..2c50ffb2 100644 --- a/py/src/braintrust/integrations/langchain/callbacks.py +++ b/py/src/braintrust/integrations/langchain/callbacks.py @@ -72,6 +72,13 @@ def _provider_from_serialized(serialized: Mapping[str, Any] | None) -> str | Non return None +def _error_for_span(error: BaseException) -> str | None: + """``GeneratorExit`` means the consumer stopped a stream early, which is not a failure.""" + if isinstance(error, GeneratorExit): + return None + return str(error) + + def _resolve_name(name: str | None, serialized: Mapping[str, Any] | None, default: str) -> str: return name or (serialized or {}).get("name") or last_item((serialized or {}).get("id") or []) or default @@ -258,7 +265,7 @@ def on_llm_error( parent_run_id: UUID | None = None, **kwargs: Any, ) -> None: - self._end_span(run_id, error=str(error)) + self._end_span(run_id, error=_error_for_span(error)) self._start_times.pop(run_id, None) self._first_token_times.pop(run_id, None) self._ttft_ms.pop(run_id, None) @@ -271,7 +278,7 @@ def on_chain_error( parent_run_id: UUID | None = None, **kwargs: Any, ) -> None: - self._end_span(run_id, error=str(error)) + self._end_span(run_id, error=_error_for_span(error)) def on_tool_error( self, @@ -281,7 +288,7 @@ def on_tool_error( parent_run_id: UUID | None = None, **kwargs: Any, ) -> None: - self._end_span(run_id, error=str(error)) + self._end_span(run_id, error=_error_for_span(error)) def on_retriever_error( self, @@ -291,7 +298,7 @@ def on_retriever_error( parent_run_id: UUID | None = None, **kwargs: Any, ) -> None: - self._end_span(run_id, error=str(error)) + self._end_span(run_id, error=_error_for_span(error)) def on_agent_action( self, diff --git a/py/src/braintrust/integrations/langchain/test_callbacks.py b/py/src/braintrust/integrations/langchain/test_callbacks.py index 1cdac475..4b95fc9e 100644 --- a/py/src/braintrust/integrations/langchain/test_callbacks.py +++ b/py/src/braintrust/integrations/langchain/test_callbacks.py @@ -1,5 +1,6 @@ # pyright: reportTypedDictNotRequiredAccess=none import base64 +import contextlib import time import uuid from pathlib import Path @@ -16,6 +17,7 @@ from langchain_core.runnables import RunnableConfig, RunnableSerializable from langchain_core.tools import tool from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph from pydantic import BaseModel, Field from .helpers import ANY, assert_matches_object, expected_prompt_chain_spans, find_spans_by_attributes @@ -230,11 +232,6 @@ def test_langgraph_state_management(logger_memory_logger): test_logger, memory_logger = logger_memory_logger assert not memory_logger.pop() - try: - from langgraph.graph import END, START, StateGraph - except ImportError: - pytest.skip("langgraph not installed") - handler = BraintrustCallbackHandler(logger=test_logger) model = ChatOpenAI( model="gpt-4o-mini", @@ -1060,3 +1057,48 @@ def calculate(operation: str, a: float, b: float) -> float: ] ], ) + + +@pytest.mark.asyncio +async def test_langgraph_astream_early_exit_is_not_an_error(logger_memory_logger): + """Leaving an ``astream()`` loop early closes the stream with ``GeneratorExit``. + + LangGraph reports that via ``on_chain_error``; it is intentional cleanup, not a + failure, so the spans must not be marked as errored. + """ + test_logger, memory_logger = logger_memory_logger + assert not memory_logger.pop() + + handler = BraintrustCallbackHandler(logger=test_logger) + + def step_one(state: dict[str, str]): + return {"step": "one"} + + def step_two(state: dict[str, str]): + return {"step": "two"} + + graph = ( + StateGraph(state_schema=dict[str, str]) + .add_node("stepOne", step_one) + .add_node("stepTwo", step_two) + .add_edge(START, "stepOne") + .add_edge("stepOne", "stepTwo") + .add_edge("stepTwo", END) + .compile() + ) + + # ``aclosing`` closes the stream as soon as we stop iterating, rather than whenever + # the event loop finalizes the abandoned async generator. + async with contextlib.aclosing( + graph.astream({}, config={"callbacks": [handler]}, stream_mode="updates") + ) as stream: + async for chunk in stream: + assert "stepOne" in chunk + break + flush() + + spans = memory_logger.pop() + langgraph_spans = find_spans_by_attributes(spans, name="LangGraph") + assert len(langgraph_spans) == 1 + assert langgraph_spans[0].get("metrics", {}).get("end") is not None + assert [span.get("error") for span in spans] == [None] * len(spans) From 696c053668aaa46505222232bc4824db824419c7 Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Fri, 6 Nov 2026 11:13:09 -0500 Subject: [PATCH 2/2] test(langchain): restore guarded langgraph imports for pylint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit > 🤖 **This text was AI-generated.** The previous commit moved the `langgraph.graph` import to the top of `test_callbacks.py`. The `pylint` session in `static_checks` doesn't install langgraph, so it failed with: ``` test_callbacks.py:20:0: E0401: Unable to import 'langgraph.graph' (import-error) ``` pylint skips `import-error` when the import is inside a `try/except ImportError` block. This puts the guarded imports back inside the two langgraph tests. Refs SDK-797 --- .../integrations/langchain/test_callbacks.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/py/src/braintrust/integrations/langchain/test_callbacks.py b/py/src/braintrust/integrations/langchain/test_callbacks.py index 4b95fc9e..362c89d2 100644 --- a/py/src/braintrust/integrations/langchain/test_callbacks.py +++ b/py/src/braintrust/integrations/langchain/test_callbacks.py @@ -17,7 +17,6 @@ from langchain_core.runnables import RunnableConfig, RunnableSerializable from langchain_core.tools import tool from langchain_openai import ChatOpenAI -from langgraph.graph import END, START, StateGraph from pydantic import BaseModel, Field from .helpers import ANY, assert_matches_object, expected_prompt_chain_spans, find_spans_by_attributes @@ -232,6 +231,11 @@ def test_langgraph_state_management(logger_memory_logger): test_logger, memory_logger = logger_memory_logger assert not memory_logger.pop() + try: + from langgraph.graph import END, START, StateGraph + except ImportError: + pytest.skip("langgraph not installed") + handler = BraintrustCallbackHandler(logger=test_logger) model = ChatOpenAI( model="gpt-4o-mini", @@ -1069,6 +1073,11 @@ async def test_langgraph_astream_early_exit_is_not_an_error(logger_memory_logger test_logger, memory_logger = logger_memory_logger assert not memory_logger.pop() + try: + from langgraph.graph import END, START, StateGraph + except ImportError: + pytest.skip("langgraph not installed") + handler = BraintrustCallbackHandler(logger=test_logger) def step_one(state: dict[str, str]):