mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-02 06:25:09 +02:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
67f6ab27b8 | ||
|
|
f7fe7c6698 |
@@ -271,7 +271,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
|
||||
f"""Ready!
|
||||
- API: http://localhost:{port}
|
||||
- Docs: http://localhost:{port}/docs
|
||||
- LangSmith Debugger: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
|
||||
- LangGraph Studio: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
|
||||
"""
|
||||
)
|
||||
sys.stdout.flush()
|
||||
@@ -652,17 +652,11 @@ def dockerfile(
|
||||
help="Wait for a debugger client to connect to the debug port before starting the server",
|
||||
default=False,
|
||||
)
|
||||
@click.option(
|
||||
"--debugger-url",
|
||||
type=str,
|
||||
default=None,
|
||||
help="URL of the LangSmith Debugger instance to connect to. Defaults to https://smith.langchain.com",
|
||||
)
|
||||
@click.option(
|
||||
"--studio-url",
|
||||
type=str,
|
||||
default=None,
|
||||
help="(Deprecated: use --debugger-url instead) URL of the LangSmith Debugger instance to connect to.",
|
||||
help="URL of the LangGraph Studio instance to connect to. Defaults to https://smith.langchain.com",
|
||||
)
|
||||
@click.option(
|
||||
"--allow-blocking",
|
||||
@@ -698,21 +692,12 @@ def dev(
|
||||
no_browser: bool,
|
||||
debug_port: Optional[int],
|
||||
wait_for_client: bool,
|
||||
debugger_url: Optional[str],
|
||||
studio_url: Optional[str],
|
||||
allow_blocking: bool,
|
||||
tunnel: bool,
|
||||
server_log_level: str,
|
||||
):
|
||||
"""CLI entrypoint for running the LangGraph API server."""
|
||||
if studio_url is not None:
|
||||
click.secho(
|
||||
"Warning: --studio-url is deprecated and will be removed in a future version. "
|
||||
"Please use --debugger-url instead.",
|
||||
fg="yellow",
|
||||
)
|
||||
if debugger_url is None:
|
||||
debugger_url = studio_url
|
||||
try:
|
||||
from langgraph_api.cli import run_server # type: ignore
|
||||
except ImportError:
|
||||
@@ -776,7 +761,7 @@ def dev(
|
||||
http=config_json.get("http"),
|
||||
ui=config_json.get("ui"),
|
||||
ui_config=config_json.get("ui_config"),
|
||||
studio_url=debugger_url,
|
||||
studio_url=studio_url,
|
||||
allow_blocking=allow_blocking,
|
||||
tunnel=tunnel,
|
||||
server_level=server_log_level,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from langgraph.pregel._write import Overwrite, overwrite
|
||||
from langgraph.pregel.main import NodeBuilder, Pregel
|
||||
|
||||
__all__ = ("Pregel", "NodeBuilder")
|
||||
__all__ = ("Pregel", "NodeBuilder", "overwrite", "Overwrite")
|
||||
|
||||
@@ -71,6 +71,7 @@ from langgraph.pregel._call import get_runnable_for_task, identifier
|
||||
from langgraph.pregel._io import read_channels
|
||||
from langgraph.pregel._log import logger
|
||||
from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode
|
||||
from langgraph.pregel._write import _Overwrite
|
||||
from langgraph.runtime import DEFAULT_RUNTIME, Runtime
|
||||
from langgraph.types import (
|
||||
All,
|
||||
@@ -198,8 +199,16 @@ def local_read(
|
||||
# apply writes
|
||||
local_channels: dict[str, BaseChannel] = {}
|
||||
for k in channels:
|
||||
cc = channels[k].copy()
|
||||
cc.update(updated[k])
|
||||
if updated[k]:
|
||||
# If any overwrite is present for this channel, reflect it directly
|
||||
ow = next((v for v in updated[k] if isinstance(v, _Overwrite)), None)
|
||||
if ow is not None:
|
||||
cc = channels[k].from_checkpoint(ow.value)
|
||||
else:
|
||||
cc = channels[k].copy()
|
||||
cc.update(updated[k])
|
||||
else:
|
||||
cc = channels[k].copy()
|
||||
local_channels[k] = cc
|
||||
# read fresh values
|
||||
values = read_channels(local_channels, select)
|
||||
@@ -277,12 +286,16 @@ def apply_writes(
|
||||
|
||||
# Group writes by channel
|
||||
pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
|
||||
overwrite_by_channel: dict[str, Any] = {}
|
||||
for task in tasks:
|
||||
for chan, val in task.writes:
|
||||
if chan in (NO_WRITES, PUSH, RESUME, INTERRUPT, RETURN, ERROR):
|
||||
pass
|
||||
elif chan in channels:
|
||||
pending_writes_by_channel[chan].append(val)
|
||||
continue
|
||||
if chan in channels:
|
||||
if isinstance(val, _Overwrite):
|
||||
overwrite_by_channel[chan] = val.value
|
||||
else:
|
||||
pending_writes_by_channel[chan].append(val)
|
||||
else:
|
||||
logger.warning(
|
||||
f"Task {task.name} with path {task.path} wrote to unknown channel {chan}, ignoring it."
|
||||
@@ -290,13 +303,26 @@ def apply_writes(
|
||||
|
||||
# Apply writes to channels
|
||||
updated_channels: set[str] = set()
|
||||
for chan, vals in pending_writes_by_channel.items():
|
||||
for chan in set(pending_writes_by_channel.keys()) | set(
|
||||
overwrite_by_channel.keys()
|
||||
):
|
||||
if chan in channels:
|
||||
if channels[chan].update(vals) and next_version is not None:
|
||||
checkpoint["channel_versions"][chan] = next_version
|
||||
# unavailable channels can't trigger tasks, so don't add them
|
||||
if chan in overwrite_by_channel:
|
||||
# Overwrite the entire channel value, bypassing reducers.
|
||||
channels[chan] = channels[chan].from_checkpoint(
|
||||
overwrite_by_channel[chan]
|
||||
)
|
||||
if next_version is not None:
|
||||
checkpoint["channel_versions"][chan] = next_version
|
||||
if channels[chan].is_available():
|
||||
updated_channels.add(chan)
|
||||
else:
|
||||
vals = pending_writes_by_channel.get(chan, [])
|
||||
if channels[chan].update(vals) and next_version is not None:
|
||||
checkpoint["channel_versions"][chan] = next_version
|
||||
# unavailable channels can't trigger tasks, so don't add them
|
||||
if channels[chan].is_available():
|
||||
updated_channels.add(chan)
|
||||
|
||||
# Channels that weren't updated in this step are notified of a new step
|
||||
if bump_step:
|
||||
|
||||
@@ -26,6 +26,32 @@ SKIP_WRITE = object()
|
||||
PASSTHROUGH = object()
|
||||
|
||||
|
||||
class _Overwrite:
|
||||
"""Marker wrapper indicating a direct channel overwrite.
|
||||
|
||||
Use via `overwrite(channel, value)` or `Overwrite(value)`.
|
||||
"""
|
||||
|
||||
__slots__ = ("value",)
|
||||
|
||||
def __init__(self, value: Any):
|
||||
self.value = value
|
||||
|
||||
|
||||
def Overwrite(value: Any) -> _Overwrite:
|
||||
"""Wrap a value to force overwrite a channel, bypassing reducers."""
|
||||
return _Overwrite(value)
|
||||
|
||||
|
||||
def overwrite(channel: str, value: Any) -> ChannelWriteEntry:
|
||||
"""Convenience factory for a write that overwrites the target channel.
|
||||
|
||||
Example:
|
||||
NodeBuilder().write_to(overwrite("foo", 123))
|
||||
"""
|
||||
return ChannelWriteEntry(channel, Overwrite(value))
|
||||
|
||||
|
||||
class ChannelWriteEntry(NamedTuple):
|
||||
channel: str
|
||||
"""Channel name to write to."""
|
||||
|
||||
Reference in New Issue
Block a user