diff --git a/docs/_scripts/copy_notebooks.py b/docs/_scripts/copy_notebooks.py index 9cb247f96..9fe8f6d03 100644 --- a/docs/_scripts/copy_notebooks.py +++ b/docs/_scripts/copy_notebooks.py @@ -42,7 +42,7 @@ _MANUAL = { "introduction.ipynb", "customer-support/customer-support.ipynb", "tutorials/tnt-llm/tnt-llm.ipynb", - "tutorials/sql-agent.ipynb" + "tutorials/sql-agent.ipynb", ], } _MANUAL_INVERSE = {v: docs_dir / k for k, vs in _MANUAL.items() for v in vs} @@ -103,9 +103,9 @@ def copy_notebooks(): continue if any(path in _HOW_TOS for path in root.split(os.sep)): dst_dir = how_tos_dir - elif 'sdk' in root.split(os.sep): + elif "sdk" in root.split(os.sep): dst_dir = cloud_sdk_dir - elif 'cloud_examples' in root.split(os.sep): + elif "cloud_examples" in root.split(os.sep): dst_dir = cloud_how_tos_dir else: dst_dir = tutorials_dir @@ -145,7 +145,7 @@ def copy_notebooks(): with open(dst_path, "w") as f: f.write(content) dst_dir = dst_dir_ - + # Top level notebooks are "how-to's" # for file in examples_dir.iterdir(): # if file.suffix.endswith(".ipynb") and not os.path.isdir( @@ -155,6 +155,7 @@ def copy_notebooks(): # dst_path = os.path.join(docs_dir, "how-tos", file.name) # shutil.copy(src_path, dst_path) + if __name__ == "__main__": clean_notebooks() copy_notebooks() diff --git a/docs/docs/reference/graphs.md b/docs/docs/reference/graphs.md index 59ec3d7b5..6a2dfd613 100644 --- a/docs/docs/reference/graphs.md +++ b/docs/docs/reference/graphs.md @@ -13,17 +13,19 @@ graph = StateGraph(MyState) ``` ::: langgraph.graph.StateGraph - handler: python +handler: python ## MessageGraph ::: langgraph.graph.message.MessageGraph - ## CompiledGraph ::: langgraph.graph.graph.CompiledGraph - handler: python + +## StreamMode + +::: langgraph.pregel.StreamMode ## Constants @@ -63,5 +65,4 @@ builder.add_conditional_edges("my_node", my_condition) ## Send -::: langgraph.constants.Send - handler: python \ No newline at end of file +::: langgraph.constants.Send \ No newline at end of file diff --git a/libs/langgraph/langgraph/checkpoint/memory.py b/libs/langgraph/langgraph/checkpoint/memory.py index d94eb3d64..5f87aa736 100644 --- a/libs/langgraph/langgraph/checkpoint/memory.py +++ b/libs/langgraph/langgraph/checkpoint/memory.py @@ -98,7 +98,7 @@ class MemorySaver(BaseCheckpointSaver): """List checkpoints from the in-memory storage. This method retrieves a list of checkpoint tuples from the in-memory storage based - on the provided config. The checkpoints are ordered by timestamp in descending order. + on the provided config. The checkpoints are ordered by timestamp in insertion order. Args: config (RunnableConfig): The config to use for listing the checkpoints. diff --git a/libs/langgraph/langgraph/graph/graph.py b/libs/langgraph/langgraph/graph/graph.py index e8c281bdc..212987e67 100644 --- a/libs/langgraph/langgraph/graph/graph.py +++ b/libs/langgraph/langgraph/graph/graph.py @@ -224,6 +224,8 @@ class Graph: def set_entry_point(self, key: str) -> None: """Specifies the first node to be called in the graph. + Equivalent to calling `add_edge(START, key)`. + Parameters: key (str): The key of the node to set as the entry point. diff --git a/libs/langgraph/langgraph/graph/message.py b/libs/langgraph/langgraph/graph/message.py index 6e2a1729e..e676deb30 100644 --- a/libs/langgraph/langgraph/graph/message.py +++ b/libs/langgraph/langgraph/graph/message.py @@ -103,8 +103,9 @@ class MessageGraph(StateGraph): >>> builder.set_finish_point("chatbot") >>> builder.compile().invoke([("user", "Hi there.")]) [HumanMessage(content="Hi there.", id='...'), AIMessage(content="Hello!", id='...')] + ``` - + ```pycon >>> from langchain_core.messages import AIMessage, HumanMessage, ToolMessage >>> from langgraph.graph.message import MessageGraph ... diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 78ae098f6..f34315966 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -160,11 +160,15 @@ class Channel: return PregelNode( channels=cast( Union[Mapping[None, str], Mapping[str, str]], - {key: channels} - if isinstance(channels, str) and key is not None - else [channels] - if isinstance(channels, str) - else {chan: chan for chan in channels}, + ( + {key: channels} + if isinstance(channels, str) and key is not None + else ( + [channels] + if isinstance(channels, str) + else {chan: chan for chan in channels} + ) + ), ), triggers=[channels] if isinstance(channels, str) else channels, tags=tags, @@ -180,15 +184,24 @@ class Channel: return ChannelWrite( [ChannelWriteEntry(c) for c in channels] + [ - ChannelWriteEntry(k, skip_none=True, mapper=coerce_to_runnable(v)) - if isinstance(v, Runnable) or callable(v) - else ChannelWriteEntry(k, value=v) + ( + ChannelWriteEntry(k, skip_none=True, mapper=coerce_to_runnable(v)) + if isinstance(v, Runnable) or callable(v) + else ChannelWriteEntry(k, value=v) + ) for k, v in kwargs.items() ] ) StreamMode = Literal["values", "updates", "debug"] +"""How the stream method should emit outputs. + +- 'values': Emit all values of the state for each step. +- 'updates': Emit only the node name(s) and updates + that were returned by the node(s) **after** each step. +- 'debug': Emit debug events for each step. +""" class Pregel( @@ -738,7 +751,73 @@ class Pregel( interrupt_after: Optional[Union[All, Sequence[str]]] = None, debug: Optional[bool] = None, ) -> Iterator[Union[dict[str, Any], Any]]: - """Stream graph steps for a single input.""" + """Stream graph steps for a single input. + + Args: + input: The input to the graph. + config: The configuration to use for the run. + stream_mode: The mode to stream output, defaults to 'values'. + Options are 'values', 'updates', and 'debug'. + values: Emit the current values of the state for each step. + updates: Emit only the updates to the state for each step. + Output is a dict with the node name as key and the updated values as value. + debug: Emit debug events for each step. + output_keys: The keys to stream, defaults to all non-context channels. + input_keys: The keys to use from the input, defaults to all input channels. + interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph. + interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph. + debug: Whether to print debug information during execution, defaults to False. + + Yields: + The output of each step in the graph. The output shape depends on the stream_mode. + + Examples: + Using different stream modes with a graph: + ```pycon + >>> import operator + >>> from typing_extensions import Annotated, TypedDict + >>> from langgraph.graph import StateGraph + >>> from libs.langgraph.langgraph.constants import START + ... + >>> class State(TypedDict): + ... alist: Annotated[list, operator.add] + ... another_list: Annotated[list, operator.add] + ... + >>> builder = StateGraph(State) + >>> builder.add_node("a", lambda _state: {"another_list": ["hi"]}) + >>> builder.add_node("b", lambda _state: {"alist": ["there"]}) + >>> builder.add_edge("a", "b") + >>> builder.add_edge(START, "a") + >>> graph = builder.compile() + ``` + With stream_mode="values": + + ```pycon + >>> for event in graph.stream({"alist": ['Ex for stream_mode="values"']}, stream_mode="values"): + ... print(event) + {'alist': ['Ex for stream_mode="values"'], 'another_list': []} + {'alist': ['Ex for stream_mode="values"'], 'another_list': ['hi']} + {'alist': ['Ex for stream_mode="values"', 'there'], 'another_list': ['hi']} + ``` + With stream_mode="updates": + + ```pycon + >>> for event in graph.stream({"alist": ['Ex for stream_mode="updates"']}, stream_mode="updates"): + ... print(event) + {'a': {'another_list': ['hi']}} + {'b': {'alist': ['there']}} + ``` + With stream_mode="debug": + + ```pycon + >>> for event in graph.stream({"alist": ['Ex for stream_mode="debug"']}, stream_mode="debug"): + ... print(event) + {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': []}, 'triggers': ['start:a']}} + {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'result': [('another_list', ['hi'])]}} + {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': ['hi']}, 'triggers': ['a']}} + {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}} + ``` + """ config = ensure_config(config) callback_manager = get_callback_manager_for_config(config) run_manager = callback_manager.on_chain_start( @@ -856,18 +935,22 @@ class Pregel( config, -1, for_execution=True, - get_next_version=self.checkpointer.get_next_version - if self.checkpointer - else _increment, + get_next_version=( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # apply input writes _apply_writes( checkpoint, channels, input_writes, - self.checkpointer.get_next_version - if self.checkpointer - else _increment, + ( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # save input checkpoint yield from put_checkpoint( @@ -904,9 +987,11 @@ class Pregel( step, for_execution=True, manager=run_manager, - get_next_version=self.checkpointer.get_next_version - if self.checkpointer - else _increment, + get_next_version=( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # if no more tasks, we're done @@ -952,9 +1037,11 @@ class Pregel( done, inflight = concurrent.futures.wait( futures, return_when=concurrent.futures.FIRST_COMPLETED, - timeout=max(0, end_time - time.monotonic()) - if end_time - else None, + timeout=( + max(0, end_time - time.monotonic()) + if end_time + else None + ), ) for fut in done: task = futures.pop(fut) @@ -1002,9 +1089,11 @@ class Pregel( checkpoint, channels, pending_writes, - self.checkpointer.get_next_version - if self.checkpointer - else _increment, + ( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # yield values output @@ -1020,14 +1109,14 @@ class Pregel( { "source": "loop", "step": step, - "writes": single( - map_output_updates(output_keys, next_tasks) - ) - if self.stream_mode == "updates" - else single( - map_output_values( - output_keys, pending_writes, channels - ), + "writes": ( + single(map_output_updates(output_keys, next_tasks)) + if self.stream_mode == "updates" + else single( + map_output_values( + output_keys, pending_writes, channels + ), + ) ), } ) @@ -1206,18 +1295,22 @@ class Pregel( config, -1, for_execution=True, - get_next_version=self.checkpointer.get_next_version - if self.checkpointer - else _increment, + get_next_version=( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # apply input writes _apply_writes( checkpoint, channels, input_writes, - self.checkpointer.get_next_version - if self.checkpointer - else _increment, + ( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # save input checkpoint for chunk in put_checkpoint( @@ -1251,9 +1344,11 @@ class Pregel( step, for_execution=True, manager=run_manager, - get_next_version=self.checkpointer.get_next_version - if self.checkpointer - else _increment, + get_next_version=( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # if no more tasks, we're done @@ -1300,9 +1395,9 @@ class Pregel( done, inflight = await asyncio.wait( futures, return_when=asyncio.FIRST_COMPLETED, - timeout=max(0, end_time - loop.time()) - if end_time - else None, + timeout=( + max(0, end_time - loop.time()) if end_time else None + ), ) for fut in done: task = futures.pop(fut) @@ -1352,9 +1447,11 @@ class Pregel( checkpoint, channels, pending_writes, - self.checkpointer.get_next_version - if self.checkpointer - else _increment, + ( + self.checkpointer.get_next_version + if self.checkpointer + else _increment + ), ) # yield current values @@ -1371,12 +1468,14 @@ class Pregel( { "source": "loop", "step": step, - "writes": single( - map_output_updates(output_keys, next_tasks) - ) - if self.stream_mode == "updates" - else single( - map_output_values(output_keys, pending_writes, channels) + "writes": ( + single(map_output_updates(output_keys, next_tasks)) + if self.stream_mode == "updates" + else single( + map_output_values( + output_keys, pending_writes, channels + ) + ) ), } ): @@ -1740,9 +1839,11 @@ def _prepare_next_tasks( }, ), run_name=packet.node, - callbacks=manager.get_child(f"graph:step:{step}") - if manager - else None, + callbacks=( + manager.get_child(f"graph:step:{step}") + if manager + else None + ), configurable={ # deque.extend is thread-safe CONFIG_KEY_SEND: partial( @@ -1820,9 +1921,11 @@ def _prepare_next_tasks( }, ), run_name=name, - callbacks=manager.get_child(f"graph:step:{step}") - if manager - else None, + callbacks=( + manager.get_child(f"graph:step:{step}") + if manager + else None + ), configurable={ # deque.extend is thread-safe CONFIG_KEY_SEND: partial(