langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

[PR #5849] perf(langgraph): Speed up method PregelLoop._emit by 243% #4685

New Issue
Closed
opened 2026-02-20 17:50:35 -05:00 by yindo · 0 comments
yindo commented 2026-02-20 17:50:35 -05:00
Owner
Copy Link
Copy Source

Original Pull Request: https://github.com/langchain-ai/langgraph/pull/5849

State: closed
Merged: No

📄 243% (2.43x) speedup for PregelLoop._emit in libs/langgraph/langgraph/pregel/loop.py

⏱️ Runtime : 1.15 milliseconds → 336 microseconds (best of 7 runs)

📝 Explanation and details

Saurabh - Creating this PR with some aggressive optimizations because this function looks important, so would benefit from being made faster.

Optimization Reasoning.

  • Attribute access: Cached self.stream, self.checkpoint_ns, self.stream.modes locally before the loop so that the loop's per-iteration cost is minimized.
  • Branching: Compute debug_remap and mode_in_stream flags and leave only essential logic in the loop.
  • Timestamp: datetime.now(timezone.utc).isoformat() is called only once per _emit() call (per debug emission batch), instead of per loop iteration.
  • Step and type logic: Precompute where possible, only evaluating "result" in v when absolutely necessary (and only if "checkpoints" is not the mode).
  • Emission: Use a single local reference, and combine structured emission into a single dict construction.
  • Logic unchanged: All side effects and return values remain exactly as before, so all behavior (including hooks to stream and debug streams) is preserved.

This restructuring should significantly decrease per-item cost of _emit, especially when emitting many values (as profiled).

✅ Correctness verification report:

Test Status
⚙️ Existing Unit Tests 🔘 None Found
🌀 Generated Regression Tests ✅ 31 Passed
⏪ Replay Tests 🔘 None Found
🔎 Concolic Coverage Tests 🔘 None Found
📊 Tests Coverage 100.0%
🌀 Generated Regression Tests and Runtime 
from __future__ import annotations

import concurrent.futures
from collections.abc import Iterator, Mapping, Sequence
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, Iterator, Literal, cast

# imports
import pytest  # used for our unit tests
from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
from langchain_core.runnables import RunnableConfig
from langgraph.cache.base import BaseCache
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import (BaseCheckpointSaver, ChannelVersions,
                                       Checkpoint, CheckpointMetadata,
                                       PendingWrite)
from langgraph.constants import (CONF, CONFIG_KEY_CHECKPOINT_ID,
                                 CONFIG_KEY_CHECKPOINT_MAP,
                                 CONFIG_KEY_CHECKPOINT_NS,
                                 CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_STREAM,
                                 CONFIG_KEY_TASK_ID, CONFIG_KEY_THREAD_ID,
                                 EMPTY_SEQ, NS_SEP)
from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
from langgraph.pregel.algo import GetNextVersion
from langgraph.pregel.executor import Submit
from langgraph.pregel.loop import PregelLoop
from langgraph.pregel.read import PregelNode
from langgraph.store.base import BaseStore
from langgraph.types import (All, CachePolicy, PregelExecutableTask,
                             PregelScratchpad, RetryPolicy, StreamChunk,
                             StreamMode, StreamProtocol)
from langgraph.utils.config import patch_configurable
from typing_extensions import ParamSpec

P = ParamSpec("P")

WritesT = Sequence[tuple[str, Any]]

def DuplexStream(*streams: StreamProtocol) -> StreamProtocol:
    def __call__(value: StreamChunk) -> None:
        for stream in streams:
            if value[1] in stream.modes:
                stream(value)

    return StreamProtocol(__call__, {mode for s in streams for mode in s.modes})
from langgraph.pregel.loop import PregelLoop


def patch_configurable(
    config: RunnableConfig | None, patch: dict[str, Any]
) -> RunnableConfig:
    if config is None:
        return {CONF: patch}
    elif CONF not in config:
        return {**config, CONF: patch}
    else:
        return {**config, CONF: {**config[CONF], **patch}}

# ------------------------
# Unit test helpers/mocks
# ------------------------

class DummyStream:
    """A dummy stream protocol for capturing emitted events."""
    def __init__(self, modes):
        self.modes = set(modes)
        self.events = []

    def __call__(self, event):
        self.events.append(event)

def dummy_values(*args, **kwargs) -> Iterator[Any]:
    """Yields a fixed sequence of values."""
    yield from ({"foo": 1}, {"bar": 2})

def empty_values(*args, **kwargs) -> Iterator[Any]:
    """Yields nothing."""
    return
    yield  # pragma: no cover

def single_value(*args, **kwargs) -> Iterator[Any]:
    yield {"result": 42}

def many_values(n):
    """Returns a generator yielding n dicts."""
    def _gen(*args, **kwargs):
        for i in range(n):
            yield {"value": i}
    return _gen

# ------------------------
# Unit tests for _emit
# ------------------------

# ---- BASIC TEST CASES ----

def make_loop(stream=None, checkpoint_ns=()):
    """Helper to create a PregelLoop with minimal config."""
    config = {CONF: {CONFIG_KEY_CHECKPOINT_NS: NS_SEP.join(checkpoint_ns) if checkpoint_ns else ""}}
    return PregelLoop(
        input=None,
        stream=stream,
        config=config,
        store=None,
        cache=None,
        checkpointer=None,
        nodes={},
        specs={},
        input_keys=[],
        output_keys=[],
        stream_keys=[],
        trigger_to_nodes={},
    )

def test_emit_basic_mode_present():
    """Test that _emit calls stream with correct mode and values when mode is present."""
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream, checkpoint_ns=("ns1", "ns2"))
    loop._emit("foo", dummy_values) # 3.23μs -> 3.42μs (5.49% slower)
    for idx, event in enumerate(stream.events):
        ns, mode, payload = event

def test_emit_basic_mode_absent():
    """Test that nothing is emitted if mode is not present in stream.modes."""
    stream = DummyStream(modes={"bar"})
    loop = make_loop(stream=stream)
    loop._emit("foo", dummy_values) # 815ns -> 903ns (9.75% slower)

def test_emit_none_stream():
    """Test that nothing happens if stream is None."""
    loop = make_loop(stream=None)
    # Should not raise
    loop._emit("foo", dummy_values) # 527ns -> 600ns (12.2% slower)

def test_emit_empty_values():
    """Test that nothing is emitted if values yields nothing."""
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream)
    loop._emit("foo", empty_values) # 1.22μs -> 1.46μs (16.3% slower)

def test_emit_single_value():
    """Test that a single value is emitted correctly."""
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream)
    loop._emit("foo", single_value) # 2.06μs -> 2.16μs (4.62% slower)
    ns, mode, payload = stream.events[0]





def test_emit_ns_empty_tuple():
    """Test that checkpoint_ns is empty tuple if not set."""
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream, checkpoint_ns=())
    loop._emit("foo", single_value) # 1.79μs -> 1.99μs (9.85% slower)
    ns, _, _ = stream.events[0]

def test_emit_ns_nonempty():
    """Test that checkpoint_ns is set properly if provided."""
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream, checkpoint_ns=("a", "b", "c"))
    loop._emit("foo", single_value) # 1.77μs -> 1.96μs (9.79% slower)
    ns, _, _ = stream.events[0]

def test_emit_non_string_mode():
    """Test that non-string mode works as long as it matches stream.modes."""
    stream = DummyStream(modes={42})
    loop = make_loop(stream=stream)
    loop._emit(42, single_value) # 1.99μs -> 2.01μs (1.39% slower)
    ns, mode, payload = stream.events[0]


def test_emit_step_affects_debug_step():
    """Test that step is used correctly in debug remap for checkpoints and tasks."""
    stream = DummyStream(modes={"debug"})
    loop = make_loop(stream=stream)
    loop.step = 10
    # checkpoints: step-1
    loop._emit("checkpoints", single_value) # 6.87μs -> 7.28μs (5.64% slower)
    dt1 = datetime.fromisoformat(stream.events[0][2]["timestamp"])
    # tasks: step
    loop.step = 11
    stream.events.clear()
    loop._emit("tasks", single_value) # 3.08μs -> 3.59μs (14.1% slower)
    dt2 = datetime.fromisoformat(stream.events[0][2]["timestamp"])

# ---- LARGE SCALE TEST CASES ----

def test_emit_large_number_of_events():
    """Test that _emit can handle a large number of events efficiently."""
    N = 500  # Keep < 1000 for performance
    stream = DummyStream(modes={"foo"})
    loop = make_loop(stream=stream)
    loop._emit("foo", many_values(N)) # 80.7μs -> 70.0μs (15.2% faster)






from __future__ import annotations

import concurrent.futures
from collections.abc import Iterator, Mapping, Sequence
from datetime import datetime, timedelta, timezone
from typing import (Any, Callable, Dict, Iterator, List, Literal, Sequence,
                    Set, Tuple, cast)

# imports
import pytest  # used for our unit tests
from langgraph.pregel.loop import PregelLoop
from typing_extensions import ParamSpec

# Dummy/Minimal versions of required classes/constants for testing
CONF = "conf"
CONFIG_KEY_CHECKPOINT_ID = "checkpoint_id"
CONFIG_KEY_CHECKPOINT_MAP = "checkpoint_map"
CONFIG_KEY_CHECKPOINT_NS = "checkpoint_ns"
CONFIG_KEY_SCRATCHPAD = "scratchpad"
CONFIG_KEY_STREAM = "stream"
CONFIG_KEY_TASK_ID = "task_id"
CONFIG_KEY_THREAD_ID = "thread_id"
EMPTY_SEQ = ()
NS_SEP = "/"

class StreamProtocol:
    def __init__(self, func: Callable[[Any], None], modes: set[str]):
        self._func = func
        self.modes = modes

    def __call__(self, value: Any) -> None:
        self._func(value)

P = ParamSpec("P")

def DuplexStream(*streams: StreamProtocol) -> StreamProtocol:
    def __call__(value: Any) -> None:
        for stream in streams:
            if value[1] in stream.modes:
                stream(value)
    return StreamProtocol(__call__, {mode for s in streams for mode in s.modes})

def patch_configurable(config: dict | None, patch: dict[str, Any]) -> dict:
    if config is None:
        return {CONF: patch}
    elif CONF not in config:
        return {**config, CONF: patch}
    else:
        return {**config, CONF: {**config[CONF], **patch}}

# Dummy classes for required types
class PregelNode: pass
class BaseStore: pass
class BaseCache(Sequence): pass
class BaseCheckpointSaver: pass
class Checkpoint: pass
class PregelScratchpad:
    def subgraph_counter(self): return 0
class RetryPolicy: pass
class CachePolicy: pass
from langgraph.pregel.loop import PregelLoop


# Helper: stream capture class
class StreamCapture:
    def __init__(self, modes: Set[str]):
        self.modes = set(modes)
        self.values: List[Any] = []
    def __call__(self, val):
        self.values.append(val)

@pytest.fixture
def dummy_config():
    return {CONF: {}}

@pytest.fixture
def dummy_nodes():
    return {}

@pytest.fixture
def dummy_specs():
    return {}

@pytest.fixture
def dummy_trigger_to_nodes():
    return {}

# --- 1. Basic Test Cases ---

Codeflash

**Original Pull Request:** https://github.com/langchain-ai/langgraph/pull/5849 **State:** closed **Merged:** No --- ### 📄 243% (2.43x) speedup for ***`PregelLoop._emit` in `libs/langgraph/langgraph/pregel/loop.py`*** ⏱️ Runtime : **`1.15 milliseconds`** **→** **`336 microseconds`** (best of `7` runs) ### 📝 Explanation and details Saurabh - Creating this PR with some aggressive optimizations because this function looks important, so would benefit from being made faster. ### Optimization Reasoning. - **Attribute access**: Cached `self.stream`, `self.checkpoint_ns`, `self.stream.modes` locally before the loop so that the loop's per-iteration cost is minimized. - **Branching**: Compute `debug_remap` and `mode_in_stream` flags and leave only essential logic in the loop. - **Timestamp**: `datetime.now(timezone.utc).isoformat()` is called only once per _emit() call (per debug emission batch), instead of per loop iteration. - **Step and type logic**: Precompute where possible, only evaluating `"result" in v` when absolutely necessary (and only if `"checkpoints"` is not the mode). - **Emission**: Use a single local reference, and combine structured emission into a single dict construction. - **Logic unchanged**: All side effects and return values remain exactly as before, so all behavior (including hooks to stream and debug streams) is preserved. This restructuring should significantly decrease per-item cost of `_emit`, especially when emitting many values (as profiled). ✅ **Correctness verification report:** | Test | Status | | --------------------------- | ----------------- | | ⚙️ Existing Unit Tests | 🔘 **None Found** | | 🌀 Generated Regression Tests | ✅ **31 Passed** | | ⏪ Replay Tests | 🔘 **None Found** | | 🔎 Concolic Coverage Tests | 🔘 **None Found** | |📊 Tests Coverage | 100.0% | <details> <summary>🌀 Generated Regression Tests and Runtime</summary> ```python from __future__ import annotations import concurrent.futures from collections.abc import Iterator, Mapping, Sequence from datetime import datetime, timedelta, timezone from typing import Any, Callable, Iterator, Literal, cast # imports import pytest # used for our unit tests from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager from langchain_core.runnables import RunnableConfig from langgraph.cache.base import BaseCache from langgraph.channels.base import BaseChannel from langgraph.checkpoint.base import (BaseCheckpointSaver, ChannelVersions, Checkpoint, CheckpointMetadata, PendingWrite) from langgraph.constants import (CONF, CONFIG_KEY_CHECKPOINT_ID, CONFIG_KEY_CHECKPOINT_MAP, CONFIG_KEY_CHECKPOINT_NS, CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_STREAM, CONFIG_KEY_TASK_ID, CONFIG_KEY_THREAD_ID, EMPTY_SEQ, NS_SEP) from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec from langgraph.pregel.algo import GetNextVersion from langgraph.pregel.executor import Submit from langgraph.pregel.loop import PregelLoop from langgraph.pregel.read import PregelNode from langgraph.store.base import BaseStore from langgraph.types import (All, CachePolicy, PregelExecutableTask, PregelScratchpad, RetryPolicy, StreamChunk, StreamMode, StreamProtocol) from langgraph.utils.config import patch_configurable from typing_extensions import ParamSpec P = ParamSpec("P") WritesT = Sequence[tuple[str, Any]] def DuplexStream(*streams: StreamProtocol) -> StreamProtocol: def __call__(value: StreamChunk) -> None: for stream in streams: if value[1] in stream.modes: stream(value) return StreamProtocol(__call__, {mode for s in streams for mode in s.modes}) from langgraph.pregel.loop import PregelLoop def patch_configurable( config: RunnableConfig | None, patch: dict[str, Any] ) -> RunnableConfig: if config is None: return {CONF: patch} elif CONF not in config: return {**config, CONF: patch} else: return {**config, CONF: {**config[CONF], **patch}} # ------------------------ # Unit test helpers/mocks # ------------------------ class DummyStream: """A dummy stream protocol for capturing emitted events.""" def __init__(self, modes): self.modes = set(modes) self.events = [] def __call__(self, event): self.events.append(event) def dummy_values(*args, **kwargs) -> Iterator[Any]: """Yields a fixed sequence of values.""" yield from ({"foo": 1}, {"bar": 2}) def empty_values(*args, **kwargs) -> Iterator[Any]: """Yields nothing.""" return yield # pragma: no cover def single_value(*args, **kwargs) -> Iterator[Any]: yield {"result": 42} def many_values(n): """Returns a generator yielding n dicts.""" def _gen(*args, **kwargs): for i in range(n): yield {"value": i} return _gen # ------------------------ # Unit tests for _emit # ------------------------ # ---- BASIC TEST CASES ---- def make_loop(stream=None, checkpoint_ns=()): """Helper to create a PregelLoop with minimal config.""" config = {CONF: {CONFIG_KEY_CHECKPOINT_NS: NS_SEP.join(checkpoint_ns) if checkpoint_ns else ""}} return PregelLoop( input=None, stream=stream, config=config, store=None, cache=None, checkpointer=None, nodes={}, specs={}, input_keys=[], output_keys=[], stream_keys=[], trigger_to_nodes={}, ) def test_emit_basic_mode_present(): """Test that _emit calls stream with correct mode and values when mode is present.""" stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream, checkpoint_ns=("ns1", "ns2")) loop._emit("foo", dummy_values) # 3.23μs -> 3.42μs (5.49% slower) for idx, event in enumerate(stream.events): ns, mode, payload = event def test_emit_basic_mode_absent(): """Test that nothing is emitted if mode is not present in stream.modes.""" stream = DummyStream(modes={"bar"}) loop = make_loop(stream=stream) loop._emit("foo", dummy_values) # 815ns -> 903ns (9.75% slower) def test_emit_none_stream(): """Test that nothing happens if stream is None.""" loop = make_loop(stream=None) # Should not raise loop._emit("foo", dummy_values) # 527ns -> 600ns (12.2% slower) def test_emit_empty_values(): """Test that nothing is emitted if values yields nothing.""" stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream) loop._emit("foo", empty_values) # 1.22μs -> 1.46μs (16.3% slower) def test_emit_single_value(): """Test that a single value is emitted correctly.""" stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream) loop._emit("foo", single_value) # 2.06μs -> 2.16μs (4.62% slower) ns, mode, payload = stream.events[0] def test_emit_ns_empty_tuple(): """Test that checkpoint_ns is empty tuple if not set.""" stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream, checkpoint_ns=()) loop._emit("foo", single_value) # 1.79μs -> 1.99μs (9.85% slower) ns, _, _ = stream.events[0] def test_emit_ns_nonempty(): """Test that checkpoint_ns is set properly if provided.""" stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream, checkpoint_ns=("a", "b", "c")) loop._emit("foo", single_value) # 1.77μs -> 1.96μs (9.79% slower) ns, _, _ = stream.events[0] def test_emit_non_string_mode(): """Test that non-string mode works as long as it matches stream.modes.""" stream = DummyStream(modes={42}) loop = make_loop(stream=stream) loop._emit(42, single_value) # 1.99μs -> 2.01μs (1.39% slower) ns, mode, payload = stream.events[0] def test_emit_step_affects_debug_step(): """Test that step is used correctly in debug remap for checkpoints and tasks.""" stream = DummyStream(modes={"debug"}) loop = make_loop(stream=stream) loop.step = 10 # checkpoints: step-1 loop._emit("checkpoints", single_value) # 6.87μs -> 7.28μs (5.64% slower) dt1 = datetime.fromisoformat(stream.events[0][2]["timestamp"]) # tasks: step loop.step = 11 stream.events.clear() loop._emit("tasks", single_value) # 3.08μs -> 3.59μs (14.1% slower) dt2 = datetime.fromisoformat(stream.events[0][2]["timestamp"]) # ---- LARGE SCALE TEST CASES ---- def test_emit_large_number_of_events(): """Test that _emit can handle a large number of events efficiently.""" N = 500 # Keep < 1000 for performance stream = DummyStream(modes={"foo"}) loop = make_loop(stream=stream) loop._emit("foo", many_values(N)) # 80.7μs -> 70.0μs (15.2% faster) from __future__ import annotations import concurrent.futures from collections.abc import Iterator, Mapping, Sequence from datetime import datetime, timedelta, timezone from typing import (Any, Callable, Dict, Iterator, List, Literal, Sequence, Set, Tuple, cast) # imports import pytest # used for our unit tests from langgraph.pregel.loop import PregelLoop from typing_extensions import ParamSpec # Dummy/Minimal versions of required classes/constants for testing CONF = "conf" CONFIG_KEY_CHECKPOINT_ID = "checkpoint_id" CONFIG_KEY_CHECKPOINT_MAP = "checkpoint_map" CONFIG_KEY_CHECKPOINT_NS = "checkpoint_ns" CONFIG_KEY_SCRATCHPAD = "scratchpad" CONFIG_KEY_STREAM = "stream" CONFIG_KEY_TASK_ID = "task_id" CONFIG_KEY_THREAD_ID = "thread_id" EMPTY_SEQ = () NS_SEP = "/" class StreamProtocol: def __init__(self, func: Callable[[Any], None], modes: set[str]): self._func = func self.modes = modes def __call__(self, value: Any) -> None: self._func(value) P = ParamSpec("P") def DuplexStream(*streams: StreamProtocol) -> StreamProtocol: def __call__(value: Any) -> None: for stream in streams: if value[1] in stream.modes: stream(value) return StreamProtocol(__call__, {mode for s in streams for mode in s.modes}) def patch_configurable(config: dict | None, patch: dict[str, Any]) -> dict: if config is None: return {CONF: patch} elif CONF not in config: return {**config, CONF: patch} else: return {**config, CONF: {**config[CONF], **patch}} # Dummy classes for required types class PregelNode: pass class BaseStore: pass class BaseCache(Sequence): pass class BaseCheckpointSaver: pass class Checkpoint: pass class PregelScratchpad: def subgraph_counter(self): return 0 class RetryPolicy: pass class CachePolicy: pass from langgraph.pregel.loop import PregelLoop # Helper: stream capture class class StreamCapture: def __init__(self, modes: Set[str]): self.modes = set(modes) self.values: List[Any] = [] def __call__(self, val): self.values.append(val) @pytest.fixture def dummy_config(): return {CONF: {}} @pytest.fixture def dummy_nodes(): return {} @pytest.fixture def dummy_specs(): return {} @pytest.fixture def dummy_trigger_to_nodes(): return {} # --- 1. Basic Test Cases --- ``` </details> [![Codeflash](https://img.shields.io/badge/Optimized%20with-Codeflash-yellow?style=flat&color=%23ffc428&logo=data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iNDgwIiBoZWlnaHQ9ImF1dG8iIHZpZXdCb3g9IjAgMCA0ODAgMjgwIiBmaWxsPSJub25lIiB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciPgo8cGF0aCBmaWxsLXJ1bGU9ImV2ZW5vZGQiIGNsaXAtcnVsZT0iZXZlbm9kZCIgZD0iTTI4Ni43IDAuMzc4NDE4SDIwMS43NTFMNTAuOTAxIDE0OC45MTFIMTM1Ljg1MUwwLjk2MDkzOCAyODEuOTk5SDk1LjQzNTJMMjgyLjMyNCA4OS45NjE2SDE5Ni4zNDVMMjg2LjcgMC4zNzg0MThaIiBmaWxsPSIjRkZDMDQzIi8+CjxwYXRoIGZpbGwtcnVsZT0iZXZlbm9kZCIgY2xpcC1ydWxlPSJldmVub2RkIiBkPSJNMzExLjYwNyAwLjM3ODkwNkwyNTguNTc4IDU0Ljk1MjZIMzc5LjU2N0w0MzIuMzM5IDAuMzc4OTA2SDMxMS42MDdaIiBmaWxsPSIjMEIwQTBBIi8+CjxwYXRoIGZpbGwtcnVsZT0iZXZlbm9kZCIgY2xpcC1ydWxlPSJldmVub2RkIiBkPSJNMzA5LjU0NyA4OS45NjAxTDI1Ni41MTggMTQ0LjI3NkgzNzcuNTA2TDQzMC4wMjEgODkuNzAyNkgzMDkuNTQ3Vjg5Ljk2MDFaIiBmaWxsPSIjMEIwQTBBIi8+CjxwYXRoIGZpbGwtcnVsZT0iZXZlbm9kZCIgY2xpcC1ydWxlPSJldmVub2RkIiBkPSJNMjQyLjg3MyAxNjQuNjZMMTg5Ljg0NCAyMTkuMjM0SDMxMC44MzNMMzYzLjM0NyAxNjQuNjZIMjQyLjg3M1oiIGZpbGw9IiMwQjBBMEEiLz4KPC9zdmc+Cg==)](https://codeflash.ai)
yindo added the pull-request label 2026-02-20 17:50:35 -05:00
yindo closed this issue 2026-02-20 17:50:35 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#4685
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?