From 82148e9bf25d6b216e228c0102117808a92f3d38 Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Wed, 29 Jan 2025 12:07:07 -0500 Subject: [PATCH] docs: update streaming docstring for Pregel and expose in api ref (#3229) --- docs/docs/concepts/functional_api.md | 4 +- docs/docs/reference/pregel.md | 9 ++ docs/mkdocs.yml | 1 + libs/langgraph/langgraph/pregel/__init__.py | 136 ++++++++++++++++++-- libs/langgraph/langgraph/types.py | 13 +- 5 files changed, 141 insertions(+), 22 deletions(-) create mode 100644 docs/docs/reference/pregel.md diff --git a/docs/docs/concepts/functional_api.md b/docs/docs/concepts/functional_api.md index 9d58519e5..d13e1b1d8 100644 --- a/docs/docs/concepts/functional_api.md +++ b/docs/docs/concepts/functional_api.md @@ -146,7 +146,7 @@ An **entrypoint** is defined by decorating a function with the `@entrypoint` dec The function **must accept a single positional argument**, which serves as the workflow input. If you need to pass multiple pieces of data, use a dictionary as the input type for the first argument. -Decorating a function with an `entrypoint` produces a Pregel instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing). +Decorating a function with an `entrypoint` produces a [`Pregel`][langgraph.pregel.Pregel.stream] instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing). You will usually want to pass a **checkpointer** to the `@entrypoint` decorator to enable persistence and use features like **human-in-the-loop**. @@ -223,7 +223,7 @@ When declaring an `entrypoint`, you can request access to additional parameters ### Executing -Using the [`@entrypoint`](#entrypoint) yields a Pregel object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods. +Using the [`@entrypoint`](#entrypoint) yields a [`Pregel`][langgraph.pregel.Pregel.stream] object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods. === "Invoke" diff --git a/docs/docs/reference/pregel.md b/docs/docs/reference/pregel.md new file mode 100644 index 000000000..a7374fb7a --- /dev/null +++ b/docs/docs/reference/pregel.md @@ -0,0 +1,9 @@ +::: langgraph.pregel.Pregel + options: + members: + - stream + - astream + - invoke + - ainvoke + - update_state + - aupdate_state diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 93451afbe..1503e4a2d 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -379,6 +379,7 @@ nav: - Errors: reference/errors.md - Types: reference/types.md - Constants: reference/constants.md + - Pregel: reference/pregel.md - Functional API: reference/func.md - LangGraph Platform: - Server API: "cloud/reference/api/api_ref.md" diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index bac334e76..a30d2fe58 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -1490,11 +1490,15 @@ class Pregel(PregelProtocol): input: The input to the graph. config: The configuration to use for the run. stream_mode: The mode to stream output, defaults to self.stream_mode. - 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. + Options are: + + - `"values"`: Emit all values in the state after each step. + When used with functional API, values are emitted once at the end of the workflow. + - `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step. + If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately. + - `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`. + - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. + - `"debug"`: Emit debug events with as much information as possible for each step. output_keys: The keys to stream, defaults to all non-context 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. @@ -1509,8 +1513,7 @@ class Pregel(PregelProtocol): ```pycon >>> import operator >>> from typing_extensions import Annotated, TypedDict - >>> from langgraph.graph import StateGraph - >>> from langgraph.constants import START + >>> from langgraph.graph import StateGraph, START ... >>> class State(TypedDict): ... alist: Annotated[list, operator.add] @@ -1550,6 +1553,57 @@ class Pregel(PregelProtocol): {'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'])]}} ``` + + With stream_mode="custom": + + ```pycon + >>> from langgraph.types import StreamWriter + ... + >>> def node_a(state: State, writer: StreamWriter): + ... writer({"custom_data": "foo"}) + ... return {"alist": ["hi"]} + ... + >>> builder = StateGraph(State) + >>> builder.add_node("a", node_a) + >>> builder.add_edge(START, "a") + >>> graph = builder.compile() + ... + >>> for event in graph.stream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"): + ... print(event) + {'custom_data': 'foo'} + ``` + + With stream_mode="messages": + + ```pycon + >>> from typing_extensions import Annotated, TypedDict + >>> from langgraph.graph import StateGraph, START + >>> from langchain_openai import ChatOpenAI + ... + >>> llm = ChatOpenAI(model="gpt-4o-mini") + ... + >>> class State(TypedDict): + ... question: str + ... answer: str + ... + >>> def node_a(state: State): + ... response = llm.invoke(state["question"]) + ... return {"answer": response.content} + ... + >>> builder = StateGraph(State) + >>> builder.add_node("a", node_a) + >>> builder.add_edge(START, "a") + >>> graph = builder.compile() + + >>> for event in graph.stream({"question": "What is the capital of France?"}, stream_mode="messages"): + ... print(event) + (AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7}) + (AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...}) + (AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...}) + ``` """ stream = SyncQueue() @@ -1712,11 +1766,15 @@ class Pregel(PregelProtocol): input: The input to the graph. config: The configuration to use for the run. stream_mode: The mode to stream output, defaults to self.stream_mode. - 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. + Options are: + + - `"values"`: Emit all values in the state after each step. + When used with functional API, values are emitted once at the end of the workflow. + - `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step. + If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately. + - `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`. + - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. + - `"debug"`: Emit debug events with as much information as possible for each step. output_keys: The keys to stream, defaults to all non-context 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. @@ -1731,8 +1789,7 @@ class Pregel(PregelProtocol): ```pycon >>> import operator >>> from typing_extensions import Annotated, TypedDict - >>> from langgraph.graph import StateGraph - >>> from langgraph.constants import START + >>> from langgraph.graph import StateGraph, START ... >>> class State(TypedDict): ... alist: Annotated[list, operator.add] @@ -1772,6 +1829,57 @@ class Pregel(PregelProtocol): {'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'])]}} ``` + + With stream_mode="custom": + + ```pycon + >>> from langgraph.types import StreamWriter + ... + >>> async def node_a(state: State, writer: StreamWriter): + ... writer({"custom_data": "foo"}) + ... return {"alist": ["hi"]} + ... + >>> builder = StateGraph(State) + >>> builder.add_node("a", node_a) + >>> builder.add_edge(START, "a") + >>> graph = builder.compile() + ... + >>> async for event in graph.astream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"): + ... print(event) + {'custom_data': 'foo'} + ``` + + With stream_mode="messages": + + ```pycon + >>> from typing_extensions import Annotated, TypedDict + >>> from langgraph.graph import StateGraph, START + >>> from langchain_openai import ChatOpenAI + ... + >>> llm = ChatOpenAI(model="gpt-4o-mini") + ... + >>> class State(TypedDict): + ... question: str + ... answer: str + ... + >>> async def node_a(state: State): + ... response = await llm.ainvoke(state["question"]) + ... return {"answer": response.content} + ... + >>> builder = StateGraph(State) + >>> builder.add_node("a", node_a) + >>> builder.add_edge(START, "a") + >>> graph = builder.compile() + + >>> for event in graph.stream({"question": "What is the capital of France?"}, stream_mode="messages"): + ... print(event) + (AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7}) + (AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...}) + (AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...}) + (AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...}) + ``` """ stream = AsyncQueue() diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 369e2c2f1..68bb9830e 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -48,12 +48,13 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver] StreamMode = Literal["values", "updates", "debug", "messages", "custom"] """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. -- 'messages': Emit LLM messages token-by-token. -- 'custom': Emit custom output `write: StreamWriter` kwarg of each node. +- `"values"`: Emit all values in the state after each step. + When used with functional API, values are emitted once at the end of the workflow. +- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step. + If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately. +- `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`. +- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. +- `"debug"`: Emit debug events with as much information as possible for each step. """ StreamWriter = Callable[[Any], None]